ποΈGitΠ―ΡΠ°ποΈ
Commit e85300531e124df08788c1b765d50af1bf6d516d
Parents : 962c619
Author : James Rich <2199651+jamesarich@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-04-11T23:22:18-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-04-12T04:22:18Z
refactor(transport): complete transport architecture overhaul β extract callback, wire BleReconnectPolicy, fix safety issues (#5080)
Changes
72 files changed, 2346 insertions(+), 2180 deletions(-)
Diff
diff --git a/core/ble/src/androidMain/kotlin/org/meshtastic/core/ble/AndroidBluetoothRepository.kt b/core/ble/src/androidMain/kotlin/org/meshtastic/core/ble/AndroidBluetoothRepository.kt
index 5b17e264be..b330453e17 100644
--- a/core/ble/src/androidMain/kotlin/org/meshtastic/core/ble/AndroidBluetoothRepository.kt
+++ b/core/ble/src/androidMain/kotlin/org/meshtastic/core/ble/AndroidBluetoothRepository.kt
@@ -31,6 +31,7 @@ import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.launch
+import kotlinx.coroutines.suspendCancellableCoroutine
import org.koin.core.annotation.Named
import org.koin.core.annotation.Single
import org.meshtastic.core.di.CoroutineDispatchers
@@ -86,7 +87,7 @@ class AndroidBluetoothRepository(
return
}
- kotlinx.coroutines.suspendCancellableCoroutine<Unit> { cont ->
+ suspendCancellableCoroutine<Unit> { cont ->
val receiver =
object : android.content.BroadcastReceiver() {
@SuppressLint("MissingPermission")
diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt
index dde1955a5e..f658d234c0 100644
--- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt
+++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt
@@ -87,7 +87,7 @@ class KableBleService(private val peripheral: Peripheral, private val serviceUui
*
* Connection attempts follow Kable's recommended pattern from the SensorTag sample: try a direct connect first, then
* fall back to `autoConnect = true` on failure. Only two attempts are made per [connect] call β the caller
- * ([BleRadioInterface]) owns the macro-level retry/backoff loop.
+ * ([BleRadioTransport]) owns the macro-level retry/backoff loop.
*/
class KableBleConnection(private val scope: CoroutineScope) : BleConnection {
diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/MeshtasticBleDevice.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/MeshtasticBleDevice.kt
index eb2ee2129f..3342cf24f0 100644
--- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/MeshtasticBleDevice.kt
+++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/MeshtasticBleDevice.kt
@@ -17,6 +17,7 @@
package org.meshtastic.core.ble
import com.juul.kable.Advertisement
+import com.juul.kable.ExperimentalApi
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
@@ -47,7 +48,7 @@ class MeshtasticBleDevice(
override val isConnected: Boolean
get() = _state.value is BleConnectionState.Connected || ActiveBleConnection.active?.address == address
- @OptIn(com.juul.kable.ExperimentalApi::class)
+ @OptIn(ExperimentalApi::class)
override suspend fun readRssi(): Int {
val active = ActiveBleConnection.active
return if (active != null && active.address == address) {
diff --git a/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfileTest.kt b/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfileTest.kt
index 8068c9387a..64286fd70f 100644
--- a/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfileTest.kt
+++ b/core/ble/src/commonTest/kotlin/org/meshtastic/core/ble/KableMeshtasticRadioProfileTest.kt
@@ -18,6 +18,7 @@ package org.meshtastic.core.ble
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.async
+import kotlinx.coroutines.flow.emptyFlow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.advanceUntilIdle
@@ -118,8 +119,8 @@ class KableMeshtasticRadioProfileTest {
fun `MeshtasticRadioProfile default awaitSubscriptionReady returns immediately`() = runTest {
val profile =
object : MeshtasticRadioProfile {
- override val fromRadio = kotlinx.coroutines.flow.emptyFlow<ByteArray>()
- override val logRadio = kotlinx.coroutines.flow.emptyFlow<ByteArray>()
+ override val fromRadio = emptyFlow<ByteArray>()
+ override val logRadio = emptyFlow<ByteArray>()
override suspend fun sendToRadio(packet: ByteArray) {}
}
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/CommandSenderImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/CommandSenderImpl.kt
index ca22f927d7..fd72ef9c71 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/CommandSenderImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/CommandSenderImpl.kt
@@ -39,18 +39,26 @@ import org.meshtastic.core.repository.PacketHandler
import org.meshtastic.core.repository.RadioConfigRepository
import org.meshtastic.core.repository.TracerouteHandler
import org.meshtastic.proto.AdminMessage
+import org.meshtastic.proto.AirQualityMetrics
import org.meshtastic.proto.ChannelSet
import org.meshtastic.proto.Constants
import org.meshtastic.proto.Data
+import org.meshtastic.proto.DeviceMetrics
+import org.meshtastic.proto.EnvironmentMetrics
+import org.meshtastic.proto.HostMetrics
import org.meshtastic.proto.LocalConfig
+import org.meshtastic.proto.LocalStats
import org.meshtastic.proto.MeshPacket
import org.meshtastic.proto.Neighbor
import org.meshtastic.proto.NeighborInfo
+import org.meshtastic.proto.Paxcount
import org.meshtastic.proto.PortNum
+import org.meshtastic.proto.PowerMetrics
import org.meshtastic.proto.Telemetry
import kotlin.math.absoluteValue
import kotlin.random.Random
import kotlin.time.Duration.Companion.hours
+import org.meshtastic.proto.Position as ProtoPosition
@Suppress("TooManyFunctions", "CyclomaticComplexMethod")
@Single
@@ -68,10 +76,6 @@ class CommandSenderImpl(
private val localConfig = MutableStateFlow(LocalConfig())
private val channelSet = MutableStateFlow(ChannelSet())
- // We'll need a way to track connection state in shared code,
- // maybe via ServiceRepository or similar.
- // For now I'll assume it's injected or available.
-
init {
radioConfigRepository.localConfigFlow.onEach { localConfig.value = it }.launchIn(scope)
radioConfigRepository.channelSetFlow.onEach { channelSet.value = it }.launchIn(scope)
@@ -141,14 +145,11 @@ class CommandSenderImpl(
if (!Data.ADAPTER.isWithinSizeLimit(data, Constants.DATA_PAYLOAD_LEN.value)) {
val actualSize = Data.ADAPTER.encodedSize(data)
p.status = MessageStatus.ERROR
- // throw RemoteException("Message too long: $actualSize bytes (max ${Constants.DATA_PAYLOAD_LEN.value})")
- // RemoteException is Android specific. For KMP we might want a custom exception.
error("Message too long: $actualSize bytes")
} else {
p.status = MessageStatus.QUEUED
}
- // TODO: Check connection state
sendNow(p)
}
@@ -191,7 +192,7 @@ class CommandSenderImpl(
return packetHandler.sendToRadioAndAwait(packet)
}
- override fun sendPosition(pos: org.meshtastic.proto.Position, destNum: Int?, wantResponse: Boolean) {
+ override fun sendPosition(pos: ProtoPosition, destNum: Int?, wantResponse: Boolean) {
val myNum = nodeManager.myNodeNum.value ?: return
val idNum = destNum ?: myNum
Logger.d { "Sending our position/time to=$idNum $pos" }
@@ -217,7 +218,7 @@ class CommandSenderImpl(
override fun requestPosition(destNum: Int, currentPosition: Position) {
val meshPosition =
- org.meshtastic.proto.Position(
+ ProtoPosition(
latitude_i = Position.degI(currentPosition.latitude),
longitude_i = Position.degI(currentPosition.longitude),
altitude = currentPosition.altitude,
@@ -240,7 +241,7 @@ class CommandSenderImpl(
override fun setFixedPosition(destNum: Int, pos: Position) {
val meshPos =
- org.meshtastic.proto.Position(
+ ProtoPosition(
latitude_i = Position.degI(pos.latitude),
longitude_i = Position.degI(pos.longitude),
altitude = pos.altitude,
@@ -293,21 +294,17 @@ class CommandSenderImpl(
if (type == TelemetryType.PAX) {
portNum = PortNum.PAXCOUNTER_APP
- payloadBytes = org.meshtastic.proto.Paxcount().encode().toByteString()
+ payloadBytes = Paxcount().encode().toByteString()
} else {
portNum = PortNum.TELEMETRY_APP
payloadBytes =
Telemetry(
- device_metrics =
- if (type == TelemetryType.DEVICE) org.meshtastic.proto.DeviceMetrics() else null,
- environment_metrics =
- if (type == TelemetryType.ENVIRONMENT) org.meshtastic.proto.EnvironmentMetrics() else null,
- air_quality_metrics =
- if (type == TelemetryType.AIR_QUALITY) org.meshtastic.proto.AirQualityMetrics() else null,
- power_metrics = if (type == TelemetryType.POWER) org.meshtastic.proto.PowerMetrics() else null,
- local_stats =
- if (type == TelemetryType.LOCAL_STATS) org.meshtastic.proto.LocalStats() else null,
- host_metrics = if (type == TelemetryType.HOST) org.meshtastic.proto.HostMetrics() else null,
+ device_metrics = if (type == TelemetryType.DEVICE) DeviceMetrics() else null,
+ environment_metrics = if (type == TelemetryType.ENVIRONMENT) EnvironmentMetrics() else null,
+ air_quality_metrics = if (type == TelemetryType.AIR_QUALITY) AirQualityMetrics() else null,
+ power_metrics = if (type == TelemetryType.POWER) PowerMetrics() else null,
+ local_stats = if (type == TelemetryType.LOCAL_STATS) LocalStats() else null,
+ host_metrics = if (type == TelemetryType.HOST) HostMetrics() else null,
)
.encode()
.toByteString()
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshActionHandlerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshActionHandlerImpl.kt
index 7f9e6c3fab..5fd34e02ed 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshActionHandlerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshActionHandlerImpl.kt
@@ -18,6 +18,7 @@ package org.meshtastic.core.data.manager
import co.touchlab.kermit.Logger
import kotlinx.coroutines.CoroutineScope
+import okio.ByteString
import okio.ByteString.Companion.toByteString
import org.koin.core.annotation.Named
import org.koin.core.annotation.Single
@@ -199,7 +200,7 @@ class MeshActionHandlerImpl(
commandSender.sendData(p)
serviceBroadcasts.broadcastMessageStatus(p.id, p.status ?: MessageStatus.UNKNOWN)
dataHandler.value.rememberDataPacket(p, myNodeNum, false)
- val bytes = p.bytes ?: okio.ByteString.EMPTY
+ val bytes = p.bytes ?: ByteString.EMPTY
analytics.track("data_send", DataPair("num_bytes", bytes.size), DataPair("type", p.dataType))
}
@@ -356,7 +357,7 @@ class MeshActionHandlerImpl(
override fun handleRequestRebootOta(requestId: Int, destNum: Int, mode: Int, hash: ByteArray?) {
val otaMode = OTAMode.fromValue(mode) ?: OTAMode.NO_REBOOT_OTA
val otaEvent =
- AdminMessage.OTAEvent(reboot_ota_mode = otaMode, ota_hash = hash?.toByteString() ?: okio.ByteString.EMPTY)
+ AdminMessage.OTAEvent(reboot_ota_mode = otaMode, ota_hash = hash?.toByteString() ?: ByteString.EMPTY)
commandSender.sendAdmin(destNum, requestId) { AdminMessage(ota_request = otaEvent) }
}
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
index 918f25719c..31e4f331d6 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
@@ -27,6 +27,7 @@ import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
+import okio.ByteString
import org.koin.core.annotation.Named
import org.koin.core.annotation.Single
import org.meshtastic.core.common.util.handledLaunch
@@ -221,7 +222,7 @@ class MeshConnectionManagerImpl(
private fun tearDownConnection() {
packetHandler.stopPacketQueue()
- commandSender.setSessionPasskey(okio.ByteString.EMPTY) // Prevent stale passkey on reconnect.
+ commandSender.setSessionPasskey(ByteString.EMPTY) // Prevent stale passkey on reconnect.
locationManager.stop()
mqttManager.stop()
}
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt
index 5da0448b5f..384f722d8c 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshDataHandlerImpl.kt
@@ -22,6 +22,7 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.launch
+import okio.ByteString
import org.koin.core.annotation.Named
import org.koin.core.annotation.Single
import org.meshtastic.core.common.util.handledLaunch
@@ -247,7 +248,7 @@ class MeshDataHandlerImpl(
val payload = packet.decoded?.payload ?: return
val u =
User.ADAPTER.decode(payload)
- .let { if (it.is_licensed == true) it.copy(public_key = okio.ByteString.EMPTY) else it }
+ .let { if (it.is_licensed == true) it.copy(public_key = ByteString.EMPTY) else it }
.let {
if (packet.via_mqtt == true && !it.long_name.endsWith(" (MQTT)")) {
it.copy(long_name = "${it.long_name} (MQTT)")
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshMessageProcessorImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshMessageProcessorImpl.kt
index 288ae9645b..000d0b41dc 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshMessageProcessorImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshMessageProcessorImpl.kt
@@ -69,7 +69,7 @@ class MeshMessageProcessorImpl(
@Volatile private var lastLocalNodeRefreshMs = 0L
private val earlyMutex = Mutex()
- private val earlyReceivedPackets = kotlin.collections.ArrayDeque<MeshPacket>()
+ private val earlyReceivedPackets = ArrayDeque<MeshPacket>()
private val maxEarlyPacketBuffer = 10240
override fun clearEarlyPackets() {
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/TracerouteHandlerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/TracerouteHandlerImpl.kt
index a5997208bd..5d2feb65ec 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/TracerouteHandlerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/TracerouteHandlerImpl.kt
@@ -65,7 +65,7 @@ class TracerouteHandlerImpl(
routeDiscovery.getTracerouteResponse(
getUser = { num ->
nodeManager.nodeDBbyNodeNum[num]?.let { "${it.user.long_name} (${it.user.short_name})" }
- ?: "Unknown" // TODO: Use core:resources once available in core:data
+ ?: "Unknown"
},
headerTowards = "Route towards destination:",
headerBack = "Route back to us:",
diff --git a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImplTest.kt b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImplTest.kt
index 55adf8b579..c6dfa7f43d 100644
--- a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImplTest.kt
+++ b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImplTest.kt
@@ -132,13 +132,10 @@ class MeshConnectionManagerImplTest {
scope,
)
- @AfterTest fun tearDown() {}
+ @AfterTest fun tearDown() = Unit
@Test
fun `Connected state triggers broadcast and config start`() = runTest(testDispatcher) {
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
-
manager = createManager(backgroundScope)
radioConnectionState.value = ConnectionState.Connected
advanceUntilIdle()
@@ -153,16 +150,6 @@ class MeshConnectionManagerImplTest {
@Test
fun `Disconnected state stops services`() = runTest(testDispatcher) {
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
- every { packetHandler.stopPacketQueue() } returns Unit
- every { locationManager.stop() } returns Unit
- every { mqttManager.stop() } returns Unit
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
- every { packetHandler.stopPacketQueue() } returns Unit
- every { locationManager.stop() } returns Unit
- every { mqttManager.stop() } returns Unit
every { nodeManager.nodeDBbyNodeNum } returns emptyMap()
manager = createManager(backgroundScope)
// Transition to Connected first so that Disconnected actually does something
@@ -191,11 +178,6 @@ class MeshConnectionManagerImplTest {
device = Config.DeviceConfig(role = Config.DeviceConfig.Role.CLIENT),
)
every { radioConfigRepository.localConfigFlow } returns flowOf(config)
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
- every { packetHandler.stopPacketQueue() } returns Unit
- every { locationManager.stop() } returns Unit
- every { mqttManager.stop() } returns Unit
every { nodeManager.nodeDBbyNodeNum } returns emptyMap()
manager = createManager(backgroundScope)
@@ -216,11 +198,6 @@ class MeshConnectionManagerImplTest {
// Power saving enabled
val config = LocalConfig(power = Config.PowerConfig(is_power_saving = true))
every { radioConfigRepository.localConfigFlow } returns flowOf(config)
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
- every { packetHandler.stopPacketQueue() } returns Unit
- every { locationManager.stop() } returns Unit
- every { mqttManager.stop() } returns Unit
manager = createManager(backgroundScope)
advanceUntilIdle()
@@ -280,11 +257,6 @@ class MeshConnectionManagerImplTest {
device = Config.DeviceConfig(role = Config.DeviceConfig.Role.ROUTER),
)
every { radioConfigRepository.localConfigFlow } returns flowOf(config)
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
- every { packetHandler.stopPacketQueue() } returns Unit
- every { locationManager.stop() } returns Unit
- every { mqttManager.stop() } returns Unit
every { nodeManager.nodeDBbyNodeNum } returns emptyMap()
manager = createManager(backgroundScope)
@@ -317,11 +289,6 @@ class MeshConnectionManagerImplTest {
// Power saving enabled so DeviceSleep is preserved (not mapped to Disconnected)
val config = LocalConfig(power = Config.PowerConfig(is_power_saving = true))
every { radioConfigRepository.localConfigFlow } returns flowOf(config)
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
- every { packetHandler.stopPacketQueue() } returns Unit
- every { locationManager.stop() } returns Unit
- every { mqttManager.stop() } returns Unit
every { nodeManager.nodeDBbyNodeNum } returns emptyMap()
// Record every state transition so we can verify ordering
@@ -367,11 +334,6 @@ class MeshConnectionManagerImplTest {
// Power saving enabled with a short ls_secs so the sleep timeout fires quickly
val config = LocalConfig(power = Config.PowerConfig(is_power_saving = true, ls_secs = 1))
every { radioConfigRepository.localConfigFlow } returns flowOf(config)
- every { packetHandler.sendToRadio(any<org.meshtastic.proto.ToRadio>()) } returns Unit
- every { serviceNotifications.updateServiceStateNotification(any(), any()) } returns Unit
- every { packetHandler.stopPacketQueue() } returns Unit
- every { locationManager.stop() } returns Unit
- every { mqttManager.stop() } returns Unit
every { nodeManager.nodeDBbyNodeNum } returns emptyMap()
val observed = mutableListOf<ConnectionState>()
diff --git a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt
index 0a1698c9a8..e0bda60759 100644
--- a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt
+++ b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt
@@ -21,6 +21,7 @@ import dev.mokkery.answering.returns
import dev.mokkery.every
import dev.mokkery.matcher.any
import dev.mokkery.mock
+import dev.mokkery.verify
import dev.mokkery.verifySuspend
import io.kotest.property.Arb
import io.kotest.property.arbitrary.int
@@ -84,6 +85,8 @@ class PacketHandlerImplTest {
val toRadio = ToRadio(packet = MeshPacket(id = 123))
handler.sendToRadio(toRadio)
+
+ verify { radioInterfaceService.sendToRadio(any()) }
}
@Test
@@ -93,6 +96,8 @@ class PacketHandlerImplTest {
handler.sendToRadio(packet)
testScheduler.runCurrent()
+
+ verify { radioInterfaceService.sendToRadio(any()) }
}
@Test
diff --git a/core/model/src/commonMain/kotlin/org/meshtastic/core/model/Capabilities.kt b/core/model/src/commonMain/kotlin/org/meshtastic/core/model/Capabilities.kt
index 25b9d812cd..4e02ae2a78 100644
--- a/core/model/src/commonMain/kotlin/org/meshtastic/core/model/Capabilities.kt
+++ b/core/model/src/commonMain/kotlin/org/meshtastic/core/model/Capabilities.kt
@@ -34,7 +34,7 @@ data class Capabilities(val firmwareVersion: String?, internal val forceEnableAl
/** Ability to mute notifications from specific nodes via admin messages. */
val canMuteNode = atLeast(V2_7_18)
- /** FIXME: Ability to request neighbor information from other nodes. Disabled until working better. */
+ /** Ability to request neighbor information from other nodes. Gated to [UNRELEASED] until working reliably. */
val canRequestNeighborInfo = atLeast(UNRELEASED)
/** Ability to send verified shared contacts. Supported since firmware v2.7.12. */
diff --git a/core/model/src/commonMain/kotlin/org/meshtastic/core/model/ConnectionState.kt b/core/model/src/commonMain/kotlin/org/meshtastic/core/model/ConnectionState.kt
index 505f187ea1..c8bbdadb5b 100644
--- a/core/model/src/commonMain/kotlin/org/meshtastic/core/model/ConnectionState.kt
+++ b/core/model/src/commonMain/kotlin/org/meshtastic/core/model/ConnectionState.kt
@@ -16,16 +16,16 @@
*/
package org.meshtastic.core.model
-sealed class ConnectionState {
+sealed interface ConnectionState {
/** We are disconnected from the device, and we should be trying to reconnect. */
- data object Disconnected : ConnectionState()
+ data object Disconnected : ConnectionState
/** We are currently attempting to connect to the device. */
- data object Connecting : ConnectionState()
+ data object Connecting : ConnectionState
/** We are connected to the device and communicating normally. */
- data object Connected : ConnectionState()
+ data object Connected : ConnectionState
/** The device is in a light sleep state, and we are waiting for it to wake up and reconnect to us. */
- data object DeviceSleep : ConnectionState()
+ data object DeviceSleep : ConnectionState
}
diff --git a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/AndroidRadioTransportFactory.kt b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/AndroidRadioTransportFactory.kt
index 28eb2175d2..426c6700b8 100644
--- a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/AndroidRadioTransportFactory.kt
+++ b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/AndroidRadioTransportFactory.kt
@@ -17,6 +17,7 @@
package org.meshtastic.core.network.radio
import android.content.Context
+import android.hardware.usb.UsbManager
import android.provider.Settings
import org.koin.core.annotation.Single
import org.meshtastic.core.ble.BleConnectionFactory
@@ -25,21 +26,23 @@ import org.meshtastic.core.ble.BluetoothRepository
import org.meshtastic.core.common.BuildConfigProvider
import org.meshtastic.core.di.CoroutineDispatchers
import org.meshtastic.core.model.DeviceType
+import org.meshtastic.core.model.InterfaceId
+import org.meshtastic.core.network.repository.UsbRepository
import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.core.repository.RadioTransport
import org.meshtastic.core.repository.RadioTransportFactory
/**
* Android implementation of [RadioTransportFactory]. Handles pure-KMP transports (BLE) via [BaseRadioTransportFactory]
- * while delegating legacy platform-specific connections (like USB/Serial, TCP, and Mocks) to the Android-specific
- * [InterfaceFactory].
+ * while creating platform-specific connections (TCP, USB/Serial, Mock, NOP) directly in [createPlatformTransport].
*/
@Single(binds = [RadioTransportFactory::class])
@Suppress("LongParameterList")
class AndroidRadioTransportFactory(
private val context: Context,
- private val interfaceFactory: Lazy<InterfaceFactory>,
private val buildConfigProvider: BuildConfigProvider,
+ private val usbRepository: UsbRepository,
+ private val usbManager: UsbManager,
scanner: BleScanner,
bluetoothRepository: BluetoothRepository,
connectionFactory: BleConnectionFactory,
@@ -48,13 +51,50 @@ class AndroidRadioTransportFactory(
override val supportedDeviceTypes: List<DeviceType> = listOf(DeviceType.BLE, DeviceType.TCP, DeviceType.USB)
- override fun isMockInterface(): Boolean =
+ override fun isMockTransport(): Boolean =
buildConfigProvider.isDebug || Settings.System.getString(context.contentResolver, "firebase.test.lab") == "true"
- override fun isPlatformAddressValid(address: String): Boolean = interfaceFactory.value.addressValid(address)
+ override fun isPlatformAddressValid(address: String): Boolean {
+ val interfaceId = address.firstOrNull()?.let { InterfaceId.forIdChar(it) } ?: return false
+ val rest = address.substring(1)
+ return when (interfaceId) {
+ InterfaceId.MOCK,
+ InterfaceId.NOP,
+ InterfaceId.TCP,
+ -> true
+ InterfaceId.SERIAL -> {
+ val deviceMap = usbRepository.serialDevices.value
+ val driver = deviceMap[rest] ?: deviceMap.values.firstOrNull()
+ driver != null && usbManager.hasPermission(driver.device)
+ }
+ InterfaceId.BLUETOOTH -> true // Handled by base class
+ }
+ }
override fun createPlatformTransport(address: String, service: RadioInterfaceService): RadioTransport {
- // Fallback to legacy factory for Serial, Mocks, and NOPs
- return interfaceFactory.value.createInterface(address, service)
+ val interfaceId = address.firstOrNull()?.let { InterfaceId.forIdChar(it) }
+ val rest = address.substring(1)
+
+ return when (interfaceId) {
+ InterfaceId.MOCK -> MockRadioTransport(callback = service, scope = service.serviceScope, address = rest)
+ InterfaceId.TCP ->
+ TcpRadioTransport(
+ callback = service,
+ scope = service.serviceScope,
+ dispatchers = dispatchers,
+ address = rest,
+ )
+ InterfaceId.SERIAL ->
+ SerialRadioTransport(
+ callback = service,
+ scope = service.serviceScope,
+ usbRepository = usbRepository,
+ address = rest,
+ )
+ InterfaceId.NOP,
+ null,
+ -> NopRadioTransport(rest)
+ InterfaceId.BLUETOOTH -> error("BLE addresses should be handled by BaseRadioTransportFactory")
+ }
}
}
diff --git a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/InterfaceFactory.kt b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/InterfaceFactory.kt
deleted file mode 100644
index b070ba013f..0000000000
--- a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/InterfaceFactory.kt
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-import org.meshtastic.core.model.InterfaceId
-import org.meshtastic.core.repository.RadioInterfaceService
-import org.meshtastic.core.repository.RadioTransport
-
-/**
- * Entry point for create radio backend instances given a specific address.
- *
- * This class is responsible for building and dissecting radio addresses based upon their interface type and the "rest"
- * of the address (which varies per implementation).
- */
-@Single
-class InterfaceFactory(
- private val nopInterfaceFactory: NopInterfaceFactory,
- private val mockSpec: Lazy<MockInterfaceSpec>,
- private val serialSpec: Lazy<SerialInterfaceSpec>,
- private val tcpSpec: Lazy<TCPInterfaceSpec>,
-) {
- internal val nopInterface by lazy { nopInterfaceFactory.create("") }
-
- private val specMap: Map<InterfaceId, InterfaceSpec<*>> by lazy {
- mapOf(
- InterfaceId.MOCK to mockSpec.value,
- InterfaceId.NOP to NopInterfaceSpec(nopInterfaceFactory),
- InterfaceId.SERIAL to serialSpec.value,
- InterfaceId.TCP to tcpSpec.value,
- )
- }
-
- fun toInterfaceAddress(interfaceId: InterfaceId, rest: String): String = "${interfaceId.id}$rest"
-
- fun createInterface(address: String, service: RadioInterfaceService): RadioTransport {
- val (spec, rest) = splitAddress(address)
- return spec?.createInterface(rest, service) ?: nopInterface
- }
-
- fun addressValid(address: String?): Boolean = address?.let {
- val (spec, rest) = splitAddress(it)
- spec?.addressValid(rest)
- } ?: false
-
- private fun splitAddress(address: String): Pair<InterfaceSpec<*>?, String> {
- if (address.isEmpty()) return Pair(null, "")
- val c = address[0].let { InterfaceId.forIdChar(it) }?.let { specMap[it] }
- val rest = address.substring(1)
- return Pair(c, rest)
- }
-}
diff --git a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterfaceFactory.kt b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterfaceFactory.kt
deleted file mode 100644
index f8c53313b7..0000000000
--- a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterfaceFactory.kt
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-import org.meshtastic.core.network.repository.UsbRepository
-import org.meshtastic.core.repository.RadioInterfaceService
-
-/** Factory for creating `SerialInterface` instances. */
-@Single
-class SerialInterfaceFactory(private val usbRepository: UsbRepository) {
- fun create(rest: String, service: RadioInterfaceService): SerialInterface =
- SerialInterface(service, usbRepository, rest)
-}
diff --git a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterfaceSpec.kt b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterfaceSpec.kt
deleted file mode 100644
index f510be3bb4..0000000000
--- a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterfaceSpec.kt
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import android.hardware.usb.UsbManager
-import com.hoho.android.usbserial.driver.UsbSerialDriver
-import org.koin.core.annotation.Single
-import org.meshtastic.core.network.repository.UsbRepository
-import org.meshtastic.core.repository.RadioInterfaceService
-
-/** Serial/USB interface backend implementation. */
-@Single
-class SerialInterfaceSpec(
- private val factory: SerialInterfaceFactory,
- private val usbManager: UsbManager,
- private val usbRepository: UsbRepository,
-) : InterfaceSpec<SerialInterface> {
- override fun createInterface(rest: String, service: RadioInterfaceService): SerialInterface =
- factory.create(rest, service)
-
- override fun addressValid(rest: String): Boolean {
- val driver = findSerial(rest) ?: return false
- return usbManager.hasPermission(driver.device)
- }
-
- internal fun findSerial(rest: String): UsbSerialDriver? {
- val deviceMap = usbRepository.serialDevices.value
- return deviceMap[rest] ?: deviceMap.values.firstOrNull()
- }
-}
diff --git a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterface.kt b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialRadioTransport.kt
similarity index 83%
rename from core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterface.kt
rename to core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialRadioTransport.kt
index 6c843caee6..bc3558800e 100644
--- a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialInterface.kt
+++ b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/SerialRadioTransport.kt
@@ -17,24 +17,28 @@
package org.meshtastic.core.network.radio
import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CoroutineScope
+import org.meshtastic.core.common.util.handledLaunch
import org.meshtastic.core.common.util.nowMillis
import org.meshtastic.core.network.repository.SerialConnection
import org.meshtastic.core.network.repository.SerialConnectionListener
import org.meshtastic.core.network.repository.UsbRepository
-import org.meshtastic.core.repository.RadioInterfaceService
-import org.meshtastic.proto.Heartbeat
-import org.meshtastic.proto.ToRadio
+import org.meshtastic.core.network.transport.HeartbeatSender
+import org.meshtastic.core.repository.RadioTransportCallback
import java.util.concurrent.atomic.AtomicReference
-/** An interface that assumes we are talking to a meshtastic device via USB serial */
-class SerialInterface(
- service: RadioInterfaceService,
+/** An Android USB/serial [RadioTransport] implementation. */
+class SerialRadioTransport(
+ callback: RadioTransportCallback,
+ scope: CoroutineScope,
private val usbRepository: UsbRepository,
private val address: String,
-) : StreamInterface(service) {
+) : StreamTransport(callback, scope) {
private var connRef = AtomicReference<SerialConnection?>()
- init {
+ private val heartbeatSender = HeartbeatSender(sendToRadio = ::handleSendToRadio, logTag = "Serial[$address]")
+
+ override fun start() {
connect()
}
@@ -116,14 +120,9 @@ class SerialInterface(
}
override fun keepAlive() {
- // Send a ToRadio heartbeat so the firmware resets its idle timer and responds with
- // a FromRadio queueStatus β proving the serial link is alive. Without this, the
- // serial transport has no way to detect a silently dead device (battery depleted,
- // firmware crash without the `rebooted` flag). The queueStatus response also feeds
- // into MeshMessageProcessorImpl.refreshLocalNodeLastHeard() to keep the local
- // node's lastHeard timestamp current.
- Logger.d { "[$address] Serial keepAlive β sending heartbeat" }
- handleSendToRadio(ToRadio(heartbeat = Heartbeat()).encode())
+ // Delegate to HeartbeatSender which sends a ToRadio heartbeat to prove the serial
+ // link is alive and keep the local node's lastHeard timestamp current.
+ scope.handledLaunch { heartbeatSender.sendHeartbeat() }
}
override fun sendBytes(p: ByteArray) {
diff --git a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/TCPInterfaceFactory.kt b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/TCPInterfaceFactory.kt
deleted file mode 100644
index 0032944489..0000000000
--- a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/TCPInterfaceFactory.kt
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-import org.meshtastic.core.di.CoroutineDispatchers
-import org.meshtastic.core.repository.RadioInterfaceService
-
-/** Factory for creating `TCPInterface` instances. */
-@Single
-class TCPInterfaceFactory(private val dispatchers: CoroutineDispatchers) {
- fun create(rest: String, service: RadioInterfaceService): TCPInterface = TCPInterface(service, dispatchers, rest)
-}
diff --git a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/TCPInterfaceSpec.kt b/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/TCPInterfaceSpec.kt
deleted file mode 100644
index 2539bc13c6..0000000000
--- a/core/network/src/androidMain/kotlin/org/meshtastic/core/network/radio/TCPInterfaceSpec.kt
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-import org.meshtastic.core.repository.RadioInterfaceService
-
-/** TCP interface backend implementation. */
-@Single
-class TCPInterfaceSpec(private val factory: TCPInterfaceFactory) : InterfaceSpec<TCPInterface> {
- override fun createInterface(rest: String, service: RadioInterfaceService): TCPInterface =
- factory.create(rest, service)
-}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BaseRadioTransportFactory.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BaseRadioTransportFactory.kt
index 2c5a02784d..55856abf99 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BaseRadioTransportFactory.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BaseRadioTransportFactory.kt
@@ -38,40 +38,41 @@ abstract class BaseRadioTransportFactory(
override fun isAddressValid(address: String?): Boolean {
val spec = address?.firstOrNull() ?: return false
- return spec in
- listOf(InterfaceId.TCP.id, InterfaceId.SERIAL.id, InterfaceId.BLUETOOTH.id, InterfaceId.MOCK.id) ||
- spec == '!' ||
- isPlatformAddressValid(address)
+ return when (spec) {
+ InterfaceId.TCP.id,
+ InterfaceId.SERIAL.id,
+ InterfaceId.BLUETOOTH.id,
+ InterfaceId.MOCK.id,
+ '!',
+ -> true
+ else -> isPlatformAddressValid(address)
+ }
}
protected open fun isPlatformAddressValid(address: String): Boolean = false
override fun toInterfaceAddress(interfaceId: InterfaceId, rest: String): String = "${interfaceId.id}$rest"
- override fun createTransport(address: String, service: RadioInterfaceService): RadioTransport = when {
- address.startsWith(InterfaceId.BLUETOOTH.id) -> {
- BleRadioInterface(
- serviceScope = service.serviceScope,
- scanner = scanner,
- bluetoothRepository = bluetoothRepository,
- connectionFactory = connectionFactory,
- service = service,
- address = address.removePrefix(InterfaceId.BLUETOOTH.id.toString()),
- )
- }
- address.startsWith("!") -> {
- BleRadioInterface(
- serviceScope = service.serviceScope,
- scanner = scanner,
- bluetoothRepository = bluetoothRepository,
- connectionFactory = connectionFactory,
- service = service,
- address = address.removePrefix("!"),
- )
- }
- else -> createPlatformTransport(address, service)
+ override fun createTransport(address: String, service: RadioInterfaceService): RadioTransport {
+ val transport =
+ when {
+ address.startsWith(InterfaceId.BLUETOOTH.id) || address.startsWith("!") -> {
+ val bleAddress = address.removePrefix(InterfaceId.BLUETOOTH.id.toString()).removePrefix("!")
+ BleRadioTransport(
+ scope = service.serviceScope,
+ scanner = scanner,
+ bluetoothRepository = bluetoothRepository,
+ connectionFactory = connectionFactory,
+ callback = service,
+ address = bleAddress,
+ )
+ }
+ else -> createPlatformTransport(address, service)
+ }
+ transport.start()
+ return transport
}
- /** Delegate to platform for Mock, TCP, or Serial/USB interfaces. */
+ /** Delegate to platform for Mock, TCP, or Serial/USB transports. */
protected abstract fun createPlatformTransport(address: String, service: RadioInterfaceService): RadioTransport
}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioInterface.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioTransport.kt
similarity index 52%
rename from core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioInterface.kt
rename to core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioTransport.kt
index 2eda521027..cfc84c6684 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioInterface.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleRadioTransport.kt
@@ -19,6 +19,7 @@
package org.meshtastic.core.network.radio
import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.DelicateCoroutinesApi
@@ -32,7 +33,6 @@ import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.isActive
import kotlinx.coroutines.job
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
@@ -47,54 +47,22 @@ import org.meshtastic.core.ble.BleWriteType
import org.meshtastic.core.ble.BluetoothRepository
import org.meshtastic.core.ble.DisconnectReason
import org.meshtastic.core.ble.MeshtasticBleConstants.SERVICE_UUID
+import org.meshtastic.core.ble.MeshtasticRadioProfile
import org.meshtastic.core.ble.classifyBleException
import org.meshtastic.core.ble.retryBleOperation
import org.meshtastic.core.ble.toMeshtasticRadioProfile
import org.meshtastic.core.common.util.nowMillis
import org.meshtastic.core.model.RadioNotConnectedException
-import org.meshtastic.core.repository.RadioInterfaceService
+import org.meshtastic.core.network.transport.HeartbeatSender
import org.meshtastic.core.repository.RadioTransport
-import org.meshtastic.proto.Heartbeat
-import org.meshtastic.proto.ToRadio
+import org.meshtastic.core.repository.RadioTransportCallback
import kotlin.concurrent.Volatile
-import kotlin.concurrent.atomics.AtomicInt
-import kotlin.concurrent.atomics.ExperimentalAtomicApi
-import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds
private const val SCAN_RETRY_COUNT = 3
private val SCAN_RETRY_DELAY = 1.seconds
private val CONNECTION_TIMEOUT = 15.seconds
-private const val RECONNECT_FAILURE_THRESHOLD = 3
-private val RECONNECT_BASE_DELAY = 5.seconds
-private val RECONNECT_MAX_DELAY = 60.seconds
-private const val RECONNECT_MAX_FAILURES = 10
-
-/** Settle delay before each connection attempt to let the Android BLE stack finish any pending disconnect cleanup. */
-private val SETTLE_DELAY = 1.seconds
-
-/**
- * Minimum time a BLE connection must stay up before we consider it "stable" and reset
- * [BleRadioInterface.consecutiveFailures]. Without this, a device at the edge of BLE range can repeatedly connect for a
- * fraction of a second and drop β each brief connection resets the failure counter so [RECONNECT_FAILURE_THRESHOLD] is
- * never reached, and the app never signals [ConnectionState.DeviceSleep].
- *
- * The value (5 s) is long enough that only connections that survive past the initial GATT setup are treated as genuine,
- * but short enough that normal reconnects after light-sleep still reset the counter promptly.
- */
-private val MIN_STABLE_CONNECTION = 5.seconds
-
-/**
- * Returns the reconnect backoff delay for a given consecutive failure count.
- *
- * Backoff schedule: 1 failure β 5 s 2 failures β 10 s 3 failures β 20 s 4 failures β 40 s 5+ failures β 60 s (capped)
- */
-internal fun computeReconnectBackoff(consecutiveFailures: Int): Duration {
- if (consecutiveFailures <= 0) return RECONNECT_BASE_DELAY
- val multiplier = 1 shl (consecutiveFailures - 1).coerceAtMost(4)
- return minOf(RECONNECT_BASE_DELAY * multiplier, RECONNECT_MAX_DELAY)
-}
/**
* Delay after writing a heartbeat before re-polling FROMRADIO.
@@ -117,27 +85,27 @@ private val GATT_CLEANUP_TIMEOUT = 5.seconds
* - Bonding and discovery.
* - Automatic reconnection logic.
* - MTU and connection parameter monitoring.
- * - Routing raw byte packets between the radio and [RadioInterfaceService].
+ * - Routing raw byte packets between the radio and [RadioTransportCallback].
*
- * @param serviceScope The coroutine scope to use for launching coroutines.
+ * @param scope The coroutine scope to use for launching coroutines.
* @param scanner The BLE scanner.
* @param bluetoothRepository The Bluetooth repository.
* @param connectionFactory The BLE connection factory.
- * @param service The [RadioInterfaceService] to use for handling radio events.
+ * @param callback The [RadioTransportCallback] to use for handling radio events.
* @param address The BLE address of the device to connect to.
*/
-class BleRadioInterface(
- private val serviceScope: CoroutineScope,
+class BleRadioTransport(
+ private val scope: CoroutineScope,
private val scanner: BleScanner,
private val bluetoothRepository: BluetoothRepository,
private val connectionFactory: BleConnectionFactory,
- private val service: RadioInterfaceService,
+ private val callback: RadioTransportCallback,
internal val address: String,
) : RadioTransport {
private val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
Logger.w(throwable) { "[$address] Uncaught exception in connectionScope" }
- serviceScope.launch {
+ scope.launch {
try {
bleConnection.disconnect()
} catch (e: Exception) {
@@ -145,13 +113,11 @@ class BleRadioInterface(
}
}
val (isPermanent, msg) = throwable.toDisconnectReason()
- service.onDisconnect(isPermanent, errorMessage = msg)
+ callback.onDisconnect(isPermanent, errorMessage = msg)
}
private val connectionScope: CoroutineScope =
- CoroutineScope(
- serviceScope.coroutineContext + SupervisorJob(serviceScope.coroutineContext.job) + exceptionHandler,
- )
+ CoroutineScope(scope.coroutineContext + SupervisorJob(scope.coroutineContext.job) + exceptionHandler)
private val bleConnection: BleConnection = connectionFactory.create(connectionScope, address)
private val writeMutex: Mutex = Mutex()
@@ -167,12 +133,19 @@ class BleRadioInterface(
@Volatile private var isFullyConnected = false
private var connectionJob: Job? = null
- private var consecutiveFailures = 0
-
- @OptIn(ExperimentalAtomicApi::class)
- private val heartbeatNonce = AtomicInt(0)
+ private val reconnectPolicy = BleReconnectPolicy()
+
+ private val heartbeatSender =
+ HeartbeatSender(
+ sendToRadio = ::handleSendToRadio,
+ afterHeartbeat = {
+ delay(HEARTBEAT_DRAIN_DELAY)
+ radioService?.requestDrain()
+ },
+ logTag = address,
+ )
- init {
+ override fun start() {
connect()
}
@@ -209,134 +182,104 @@ class BleRadioInterface(
throw RadioNotConnectedException("Device not found at address $address")
}
- @Suppress("LongMethod", "CyclomaticComplexMethod")
private fun connect() {
connectionJob =
connectionScope.launch {
- while (isActive) {
- try {
- // Settle delay: let the Android BLE stack finish any pending
- // disconnect cleanup before starting a new connection attempt.
- delay(SETTLE_DELAY)
-
- connectionStartTime = nowMillis
- Logger.i { "[$address] BLE connection attempt started" }
-
- val device = findDevice()
-
- // Bond before connecting: firmware may require an encrypted link,
- // and without a bond Android fails with status 5 or 133.
- // No-op on Desktop/JVM where the OS handles pairing automatically.
- if (!bluetoothRepository.isBonded(address)) {
- Logger.i { "[$address] Device not bonded, initiating bonding" }
- @Suppress("TooGenericExceptionCaught")
- try {
- bluetoothRepository.bond(device)
- Logger.i { "[$address] Bonding successful" }
- } catch (e: Exception) {
- Logger.w(e) { "[$address] Bonding failed, attempting connection anyway" }
- }
+ reconnectPolicy.execute(
+ attempt = {
+ try {
+ attemptConnection()
+ } catch (e: CancellationException) {
+ throw e
+ } catch (e: Exception) {
+ val failureTime = (nowMillis - connectionStartTime).milliseconds
+ Logger.w(e) { "[$address] Failed to connect after $failureTime" }
+ BleReconnectPolicy.Outcome.Failed(e)
}
+ },
+ onTransientDisconnect = { error ->
+ val msg = error?.toDisconnectReason()?.second ?: "Device unreachable"
+ callback.onDisconnect(isPermanent = false, errorMessage = msg)
+ },
+ onPermanentDisconnect = { error ->
+ val msg = error?.toDisconnectReason()?.second ?: "Device unreachable"
+ callback.onDisconnect(isPermanent = true, errorMessage = msg)
+ },
+ )
+ }
+ }
- val state = bleConnection.connectAndAwait(device, CONNECTION_TIMEOUT)
+ /**
+ * Performs a single BLE connect-and-wait cycle.
+ *
+ * Finds the device, bonds if needed, connects, discovers services, and waits for disconnect. Returns a
+ * [BleReconnectPolicy.Outcome] describing how the connection ended.
+ */
+ @Suppress("CyclomaticComplexMethod")
+ private suspend fun attemptConnection(): BleReconnectPolicy.Outcome {
+ connectionStartTime = nowMillis
+ Logger.i { "[$address] BLE connection attempt started" }
+
+ val device = findDevice()
+
+ // Bond before connecting: firmware may require an encrypted link,
+ // and without a bond Android fails with status 5 or 133.
+ // No-op on Desktop/JVM where the OS handles pairing automatically.
+ if (!bluetoothRepository.isBonded(address)) {
+ Logger.i { "[$address] Device not bonded, initiating bonding" }
+ @Suppress("TooGenericExceptionCaught")
+ try {
+ bluetoothRepository.bond(device)
+ Logger.i { "[$address] Bonding successful" }
+ } catch (e: Exception) {
+ Logger.w(e) { "[$address] Bonding failed, attempting connection anyway" }
+ }
+ }
- if (state !is BleConnectionState.Connected) {
- throw RadioNotConnectedException("Failed to connect to device at address $address")
- }
+ val state = bleConnection.connectAndAwait(device, CONNECTION_TIMEOUT)
- // Only reset failures if connection was stable (see MIN_STABLE_CONNECTION).
- val gattConnectedAt = nowMillis
- isFullyConnected = true
- onConnected()
-
- // Scope the connectionState listener to this iteration so it's
- // cancelled automatically before the next reconnect cycle.
- var disconnectReason: DisconnectReason = DisconnectReason.Unknown
- coroutineScope {
- bleConnection.connectionState
- .onEach { s ->
- if (s is BleConnectionState.Disconnected && isFullyConnected) {
- isFullyConnected = false
- disconnectReason = s.reason
- onDisconnected()
- }
- }
- .catch { e -> Logger.w(e) { "[$address] bleConnection.connectionState flow crashed" } }
- .launchIn(this)
-
- discoverServicesAndSetupCharacteristics()
-
- bleConnection.connectionState.first { it is BleConnectionState.Disconnected }
- }
+ if (state !is BleConnectionState.Connected) {
+ throw RadioNotConnectedException("Failed to connect to device at address $address")
+ }
- Logger.i {
- "[$address] BLE connection dropped (reason: $disconnectReason), preparing to reconnect"
- }
+ val gattConnectedAt = nowMillis
+ isFullyConnected = true
+ onConnected()
+
+ // Scope the connectionState listener to this iteration so it's
+ // cancelled automatically before the next reconnect cycle.
+ var disconnectReason: DisconnectReason = DisconnectReason.Unknown
+ coroutineScope {
+ bleConnection.connectionState
+ .onEach { s ->
+ if (s is BleConnectionState.Disconnected && isFullyConnected) {
+ isFullyConnected = false
+ disconnectReason = s.reason
+ onDisconnected()
+ }
+ }
+ .catch { e -> Logger.w(e) { "[$address] bleConnection.connectionState flow crashed" } }
+ .launchIn(this)
- // Skip failure counting for intentional disconnects.
- if (disconnectReason is DisconnectReason.LocalDisconnect) {
- consecutiveFailures = 0
- continue
- }
+ discoverServicesAndSetupCharacteristics()
- // A connection that drops almost immediately (< MIN_STABLE_CONNECTION)
- // is treated as a failure β the BLE stack may have "connected" to a
- // cached GATT profile before realising the device is gone.
- val connectionUptime = (nowMillis - gattConnectedAt).milliseconds
- if (connectionUptime >= MIN_STABLE_CONNECTION) {
- consecutiveFailures = 0
- } else {
- consecutiveFailures++
- Logger.w {
- "[$address] Connection lasted only $connectionUptime " +
- "(< $MIN_STABLE_CONNECTION) β treating as failure " +
- "(consecutive failures: $consecutiveFailures)"
- }
- if (consecutiveFailures >= RECONNECT_MAX_FAILURES) {
- Logger.e { "[$address] Giving up after $consecutiveFailures unstable connections" }
- service.onDisconnect(
- isPermanent = true,
- errorMessage = "Device unreachable (unstable connection)",
- )
- return@launch
- }
- if (consecutiveFailures >= RECONNECT_FAILURE_THRESHOLD) {
- service.onDisconnect(
- isPermanent = false,
- errorMessage = "Device unreachable (unstable connection)",
- )
- }
- }
- } catch (e: kotlinx.coroutines.CancellationException) {
- Logger.d { "[$address] BLE connection coroutine cancelled" }
- throw e
- } catch (e: Exception) {
- val failureTime = (nowMillis - connectionStartTime).milliseconds
- consecutiveFailures++
- Logger.w(e) {
- "[$address] Failed to connect to device after $failureTime " +
- "(consecutive failures: $consecutiveFailures)"
- }
+ bleConnection.connectionState.first { it is BleConnectionState.Disconnected }
+ }
- // Give up permanently to stop draining battery.
- if (consecutiveFailures >= RECONNECT_MAX_FAILURES) {
- Logger.e { "[$address] Giving up after $consecutiveFailures consecutive failures" }
- val (_, msg) = e.toDisconnectReason()
- service.onDisconnect(isPermanent = true, errorMessage = msg)
- return@launch
- }
+ Logger.i { "[$address] BLE connection dropped (reason: $disconnectReason), preparing to reconnect" }
- // Signal DeviceSleep so MeshConnectionManagerImpl starts its sleep timeout.
- if (consecutiveFailures >= RECONNECT_FAILURE_THRESHOLD) {
- handleFailure(e)
- }
+ val wasIntentional = disconnectReason is DisconnectReason.LocalDisconnect
+ val connectionUptime = (nowMillis - gattConnectedAt).milliseconds
+ val wasStable = connectionUptime >= reconnectPolicy.minStableConnection
- val backoff = computeReconnectBackoff(consecutiveFailures)
- Logger.d { "[$address] Retrying in $backoff (failure #$consecutiveFailures)" }
- delay(backoff)
- }
- }
+ if (!wasStable && !wasIntentional) {
+ Logger.w {
+ "[$address] Connection lasted only $connectionUptime " +
+ "(< ${reconnectPolicy.minStableConnection}) β treating as unstable"
}
+ }
+
+ return BleReconnectPolicy.Outcome.Disconnected(wasStable = wasStable, wasIntentional = wasIntentional)
}
private suspend fun onConnected() {
@@ -354,7 +297,7 @@ class BleRadioInterface(
radioService = null
Logger.i { "[$address] BLE disconnected - ${formatSessionStats()}" }
// Signal immediately so the UI reflects the disconnect while reconnect continues.
- service.onDisconnect(isPermanent = false)
+ callback.onDisconnect(isPermanent = false)
}
private suspend fun discoverServicesAndSetupCharacteristics() {
@@ -384,7 +327,7 @@ class BleRadioInterface(
}
.launchIn(this)
- this@BleRadioInterface.radioService = radioService
+ this@BleRadioTransport.radioService = radioService
Logger.i { "[$address] Profile service active and characteristics subscribed" }
@@ -395,7 +338,7 @@ class BleRadioInterface(
val maxLen = bleConnection.maximumWriteValueLength(BleWriteType.WITHOUT_RESPONSE)
Logger.i { "[$address] BLE Radio Session Ready. Max write length (WITHOUT_RESPONSE): $maxLen bytes" }
- this@BleRadioInterface.service.onConnect()
+ this@BleRadioTransport.callback.onConnect()
}
} catch (e: Exception) {
Logger.w(e) { "[$address] Profile service discovery or operation failed" }
@@ -409,7 +352,7 @@ class BleRadioInterface(
}
}
- @Volatile private var radioService: org.meshtastic.core.ble.MeshtasticRadioProfile? = null
+ @Volatile private var radioService: MeshtasticRadioProfile? = null
// --- RadioTransport Implementation ---
@@ -445,36 +388,19 @@ class BleRadioInterface(
}
}
- @OptIn(ExperimentalAtomicApi::class)
override fun keepAlive() {
- // Send a ToRadio heartbeat so the firmware resets its power-saving idle timer.
- // The firmware only resets the timer on writes to the TORADIO characteristic; a
- // BLE-level GATT keepalive is invisible to it. Without this the device may enter
- // light-sleep and drop the BLE connection after ~60 s of application inactivity.
- //
- // Each heartbeat uses a distinct nonce to vary the wire bytes, preventing the
- // firmware's per-connection duplicate-write filter from silently dropping it.
- val nonce = heartbeatNonce.fetchAndAdd(1)
- Logger.v { "[$address] BLE keepAlive β sending ToRadio heartbeat (nonce=$nonce)" }
- handleSendToRadio(ToRadio(heartbeat = Heartbeat(nonce = nonce)).encode())
-
- // The firmware responds to heartbeats by queuing a `queueStatus` FromRadio packet
- // on the next getFromRadio() call, but it does NOT send a FROMNUM notification for
- // it. The immediate drain trigger in sendToRadio() fires before the ESP32's async
- // task queue has processed the heartbeat, so the response sits unread. Schedule a
- // delayed re-drain to pick it up.
- connectionScope.launch {
- delay(HEARTBEAT_DRAIN_DELAY)
- radioService?.requestDrain()
- }
+ // Delegate to HeartbeatSender which sends a ToRadio heartbeat with a unique nonce
+ // so the firmware resets its power-saving idle timer. After sending, it schedules
+ // a delayed re-drain to pick up the queueStatus response.
+ connectionScope.launch { heartbeatSender.sendHeartbeat() }
}
/** Closes the connection to the device. */
override fun close() {
Logger.i { "[$address] Disconnecting. ${formatSessionStats()}" }
connectionScope.cancel("close() called")
- // GATT cleanup must outlive serviceScope cancellation β GlobalScope is intentional.
- // SharedRadioInterfaceService cancels serviceScope immediately after close(), so a
+ // GATT cleanup must outlive scope cancellation β GlobalScope is intentional.
+ // SharedRadioInterfaceService cancels the scope immediately after close(), so a
// coroutine launched there may never run, leaking BluetoothGatt (causes GATT 133).
@OptIn(DelicateCoroutinesApi::class)
GlobalScope.launch {
@@ -493,12 +419,12 @@ class BleRadioInterface(
"[$address] Dispatching packet #$packetsReceived " +
"(${packet.size} bytes, total RX: $bytesReceived bytes)"
}
- service.handleFromRadio(packet)
+ callback.handleFromRadio(packet)
}
private fun handleFailure(throwable: Throwable) {
val (isPermanent, msg) = throwable.toDisconnectReason()
- service.onDisconnect(isPermanent, errorMessage = msg)
+ callback.onDisconnect(isPermanent, errorMessage = msg)
}
/** Formats a one-line session statistics summary for logging. */
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleReconnectPolicy.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleReconnectPolicy.kt
new file mode 100644
index 0000000000..cef746af04
--- /dev/null
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/BleReconnectPolicy.kt
@@ -0,0 +1,170 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.network.radio
+
+import co.touchlab.kermit.Logger
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlin.coroutines.coroutineContext
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * Encapsulates the BLE reconnection policy with exponential backoff.
+ *
+ * The policy tracks consecutive failures and decides whether to retry, signal a transient disconnect (DeviceSleep), or
+ * give up permanently.
+ *
+ * @param maxFailures maximum consecutive failures before giving up permanently
+ * @param failureThreshold after this many consecutive failures, signal a transient disconnect
+ * @param settleDelay delay before each connection attempt to let the BLE stack settle
+ * @param minStableConnection minimum time a connection must stay up to be considered "stable"
+ * @param backoffStrategy computes the backoff delay for a given failure count
+ */
+class BleReconnectPolicy(
+ private val maxFailures: Int = DEFAULT_MAX_FAILURES,
+ private val failureThreshold: Int = DEFAULT_FAILURE_THRESHOLD,
+ private val settleDelay: Duration = DEFAULT_SETTLE_DELAY,
+ /** Minimum time a connection must stay up to be considered "stable". Exposed for callers to compare uptime. */
+ val minStableConnection: Duration = DEFAULT_MIN_STABLE_CONNECTION,
+ private val backoffStrategy: (attempt: Int) -> Duration = ::computeReconnectBackoff,
+) {
+ /** Outcome of a single reconnect iteration. */
+ sealed interface Outcome {
+ /** Connection attempt succeeded and then eventually disconnected. */
+ data class Disconnected(val wasStable: Boolean, val wasIntentional: Boolean) : Outcome
+
+ /** Connection attempt failed with an exception. */
+ data class Failed(val error: Throwable) : Outcome
+ }
+
+ /** Action the caller should take after the policy processes an outcome. */
+ sealed interface Action {
+ /** Retry the connection after the specified backoff delay. */
+ data class Retry(val backoff: Duration) : Action
+
+ /** Signal a transient disconnect to higher layers. */
+ data class SignalTransient(val backoff: Duration) : Action
+
+ /** Give up permanently. */
+ data object GiveUp : Action
+
+ /** Continue immediately (e.g. after an intentional disconnect). */
+ data object Continue : Action
+ }
+
+ internal var consecutiveFailures: Int = 0
+ private set
+
+ /** Processes the outcome of a connection attempt and returns the action the caller should take. */
+ fun processOutcome(outcome: Outcome): Action = when (outcome) {
+ is Outcome.Disconnected -> {
+ if (outcome.wasIntentional) {
+ consecutiveFailures = 0
+ Action.Continue
+ } else if (outcome.wasStable) {
+ consecutiveFailures = 0
+ Action.Continue
+ } else {
+ consecutiveFailures++
+ Logger.w { "Unstable connection (consecutive failures: $consecutiveFailures)" }
+ evaluateFailure()
+ }
+ }
+ is Outcome.Failed -> {
+ consecutiveFailures++
+ Logger.w { "Connection failed (consecutive failures: $consecutiveFailures)" }
+ evaluateFailure()
+ }
+ }
+
+ private fun evaluateFailure(): Action {
+ if (consecutiveFailures >= maxFailures) {
+ return Action.GiveUp
+ }
+ val backoff = backoffStrategy(consecutiveFailures)
+ return if (consecutiveFailures >= failureThreshold) {
+ Action.SignalTransient(backoff)
+ } else {
+ Action.Retry(backoff)
+ }
+ }
+
+ /**
+ * Runs the reconnect loop, calling [attempt] for each iteration.
+ *
+ * The [attempt] lambda should perform a single connect-and-wait cycle and return the [Outcome] when the connection
+ * drops or an error occurs.
+ *
+ * @param attempt performs a single connection attempt and returns the outcome
+ * @param onTransientDisconnect called when the policy decides to signal a transient disconnect
+ * @param onPermanentDisconnect called when the policy gives up permanently
+ */
+ suspend fun execute(
+ attempt: suspend () -> Outcome,
+ onTransientDisconnect: suspend (Throwable?) -> Unit,
+ onPermanentDisconnect: suspend (Throwable?) -> Unit,
+ ) {
+ while (coroutineContext.isActive) {
+ delay(settleDelay)
+
+ val outcome = attempt()
+ val lastError = (outcome as? Outcome.Failed)?.error
+
+ when (val action = processOutcome(outcome)) {
+ is Action.Continue -> continue
+ is Action.Retry -> {
+ Logger.d { "Retrying in ${action.backoff} (failure #$consecutiveFailures)" }
+ delay(action.backoff)
+ }
+ is Action.SignalTransient -> {
+ onTransientDisconnect(lastError)
+ Logger.d { "Retrying in ${action.backoff} (failure #$consecutiveFailures)" }
+ delay(action.backoff)
+ }
+ is Action.GiveUp -> {
+ Logger.e { "Giving up after $consecutiveFailures consecutive failures" }
+ onPermanentDisconnect(lastError)
+ return
+ }
+ }
+ }
+ }
+
+ companion object {
+ const val DEFAULT_MAX_FAILURES = 10
+ const val DEFAULT_FAILURE_THRESHOLD = 3
+ val DEFAULT_SETTLE_DELAY = 1.seconds
+ val DEFAULT_MIN_STABLE_CONNECTION = 5.seconds
+
+ internal val RECONNECT_BASE_DELAY = 5.seconds
+ internal val RECONNECT_MAX_DELAY = 60.seconds
+ internal const val BACKOFF_MAX_EXPONENT = 4
+ }
+}
+
+/**
+ * Returns the reconnect backoff delay for a given consecutive failure count.
+ *
+ * Backoff schedule: 1 failure β 5 s, 2 failures β 10 s, 3 failures β 20 s, 4 failures β 40 s, 5+ failures β 60 s
+ * (capped).
+ */
+internal fun computeReconnectBackoff(consecutiveFailures: Int): Duration {
+ if (consecutiveFailures <= 0) return BleReconnectPolicy.RECONNECT_BASE_DELAY
+ val multiplier = 1 shl (consecutiveFailures - 1).coerceAtMost(BleReconnectPolicy.BACKOFF_MAX_EXPONENT)
+ return minOf(BleReconnectPolicy.RECONNECT_BASE_DELAY * multiplier, BleReconnectPolicy.RECONNECT_MAX_DELAY)
+}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/InterfaceSpec.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/InterfaceSpec.kt
deleted file mode 100644
index aec9ec6673..0000000000
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/InterfaceSpec.kt
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.meshtastic.core.repository.RadioInterfaceService
-import org.meshtastic.core.repository.RadioTransport
-
-/** This interface defines the contract that all radio backend implementations must adhere to. */
-interface InterfaceSpec<T : RadioTransport> {
- fun createInterface(rest: String, service: RadioInterfaceService): T
-
- /** Return true if this address is still acceptable. For BLE that means, still bonded */
- fun addressValid(rest: String): Boolean = true
-}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterfaceFactory.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterfaceFactory.kt
deleted file mode 100644
index 492b5782c8..0000000000
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterfaceFactory.kt
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-import org.meshtastic.core.repository.RadioInterfaceService
-
-/** Factory for creating `MockInterface` instances. */
-@Single
-class MockInterfaceFactory {
- fun create(rest: String, service: RadioInterfaceService): MockInterface = MockInterface(service, rest)
-}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterfaceSpec.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterfaceSpec.kt
deleted file mode 100644
index 0f77cb5dc4..0000000000
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterfaceSpec.kt
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-import org.meshtastic.core.repository.RadioInterfaceService
-
-/** Mock interface backend implementation. */
-@Single
-class MockInterfaceSpec(private val factory: MockInterfaceFactory) : InterfaceSpec<MockInterface> {
- override fun createInterface(rest: String, service: RadioInterfaceService): MockInterface =
- factory.create(rest, service)
-
- /** Return true if this address is still acceptable. For BLE that means, still bonded */
- override fun addressValid(rest: String): Boolean = true
-}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterface.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockRadioTransport.kt
similarity index 90%
rename from core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterface.kt
rename to core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockRadioTransport.kt
index 4990ee7abd..78d3d4cebd 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockInterface.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/MockRadioTransport.kt
@@ -17,6 +17,7 @@
package org.meshtastic.core.network.radio
import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.delay
import okio.ByteString.Companion.encodeUtf8
import okio.ByteString.Companion.toByteString
@@ -25,8 +26,8 @@ import org.meshtastic.core.common.util.nowSeconds
import org.meshtastic.core.model.Channel
import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.util.getInitials
-import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.core.repository.RadioTransport
+import org.meshtastic.core.repository.RadioTransportCallback
import org.meshtastic.proto.AdminMessage
import org.meshtastic.proto.Config
import org.meshtastic.proto.Data
@@ -55,9 +56,13 @@ private val defaultLoRaConfig = Config.LoRaConfig(use_preset = true, region = Co
private val defaultChannel = ProtoChannel(settings = Channel.default.settings, role = ProtoChannel.Role.PRIMARY)
-/** A simulated interface that is used for testing in the simulator */
+/** A simulated transport that is used for testing in the simulator. */
@Suppress("detekt:TooManyFunctions", "detekt:MagicNumber")
-class MockInterface(private val service: RadioInterfaceService, val address: String) : RadioTransport {
+class MockRadioTransport(
+ private val callback: RadioTransportCallback,
+ private val scope: CoroutineScope,
+ val address: String,
+) : RadioTransport {
companion object {
private const val MY_NODE = 0x42424242
@@ -68,13 +73,22 @@ class MockInterface(private val service: RadioInterfaceService, val address: Str
// an infinite sequence of ints
private val packetIdSequence = generateSequence { currentPacketId++ }.iterator()
- init {
- Logger.i { "Starting the mock interface" }
- service.onConnect() // Tell clients they can use the API
+ override fun start() {
+ Logger.i { "Starting the mock transport" }
+ callback.onConnect() // Tell clients they can use the API
}
override fun handleSendToRadio(p: ByteArray) {
val pr = ToRadio.ADAPTER.decode(p)
+
+ // Intercept want_config handshake β send config response only when requested,
+ // mirroring the behaviour of real firmware which waits for want_config_id.
+ val wantConfigId = pr.want_config_id ?: 0
+ if (wantConfigId != 0) {
+ sendConfigResponse(wantConfigId)
+ return
+ }
+
val packet = pr.packet
if (packet != null) {
sendQueueStatus(packet.id)
@@ -83,11 +97,10 @@ class MockInterface(private val service: RadioInterfaceService, val address: Str
val data = packet?.decoded
when {
- (pr.want_config_id ?: 0) != 0 -> sendConfigResponse(pr.want_config_id ?: 0)
data != null && data.portnum == PortNum.ADMIN_APP ->
handleAdminPacket(pr, AdminMessage.ADAPTER.decode(data.payload))
packet != null && packet.want_ack == true -> sendFakeAck(pr)
- else -> Logger.i { "Ignoring data sent to mock interface $pr" }
+ else -> Logger.i { "Ignoring data sent to mock transport $pr" }
}
}
@@ -127,12 +140,12 @@ class MockInterface(private val service: RadioInterfaceService, val address: Str
)
}
- else -> Logger.i { "Ignoring admin sent to mock interface $d" }
+ else -> Logger.i { "Ignoring admin sent to mock transport $d" }
}
}
override fun close() {
- Logger.i { "Closing the mock interface" }
+ Logger.i { "Closing the mock transport" }
}
// / Generate a fake text message from a node
@@ -279,7 +292,7 @@ class MockInterface(private val service: RadioInterfaceService, val address: Str
Data(portnum = PortNum.ROUTING_APP, payload = Routing().encode().toByteString(), request_id = msgId),
)
- private fun sendQueueStatus(msgId: Int) = service.handleFromRadio(
+ private fun sendQueueStatus(msgId: Int) = callback.handleFromRadio(
FromRadio(queueStatus = QueueStatus(res = 0, free = 16, mesh_packet_id = msgId)).encode(),
)
@@ -291,14 +304,14 @@ class MockInterface(private val service: RadioInterfaceService, val address: Str
toIn,
Data(portnum = PortNum.ADMIN_APP, payload = adminMsg.encode().toByteString(), request_id = reqId),
)
- service.handleFromRadio(p.encode())
+ callback.handleFromRadio(p.encode())
}
// / Send a fake ack packet back if the sender asked for want_ack
- private fun sendFakeAck(pr: ToRadio) = service.serviceScope.handledLaunch {
+ private fun sendFakeAck(pr: ToRadio) = scope.handledLaunch {
val packet = pr.packet ?: return@handledLaunch
delay(2000)
- service.handleFromRadio(makeAck(MY_NODE + 1, packet.from, packet.id).encode())
+ callback.handleFromRadio(makeAck(MY_NODE + 1, packet.from, packet.id).encode())
}
private fun sendConfigResponse(configId: Int) {
@@ -353,6 +366,6 @@ class MockInterface(private val service: RadioInterfaceService, val address: Str
makeNodeStatus(MY_NODE + 1),
)
- packets.forEach { p -> service.handleFromRadio(p.encode()) }
+ packets.forEach { p -> callback.handleFromRadio(p.encode()) }
}
}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterfaceFactory.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterfaceFactory.kt
deleted file mode 100644
index 5d9991e346..0000000000
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterfaceFactory.kt
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-
-/** Factory for creating `NopInterface` instances. */
-@Single
-class NopInterfaceFactory {
- fun create(rest: String): NopInterface = NopInterface(rest)
-}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterfaceSpec.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterfaceSpec.kt
deleted file mode 100644
index df77578bfa..0000000000
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterfaceSpec.kt
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Copyright (c) 2025-2026 Meshtastic LLC
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-package org.meshtastic.core.network.radio
-
-import org.koin.core.annotation.Single
-import org.meshtastic.core.repository.RadioInterfaceService
-
-/** No-op interface backend implementation. */
-@Single
-class NopInterfaceSpec(private val factory: NopInterfaceFactory) : InterfaceSpec<NopInterface> {
- override fun createInterface(rest: String, service: RadioInterfaceService): NopInterface = factory.create(rest)
-}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterface.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopRadioTransport.kt
similarity index 69%
rename from core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterface.kt
rename to core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopRadioTransport.kt
index 27348635ca..db807081a6 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopInterface.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/NopRadioTransport.kt
@@ -18,7 +18,14 @@ package org.meshtastic.core.network.radio
import org.meshtastic.core.repository.RadioTransport
-class NopInterface(val address: String) : RadioTransport {
+/**
+ * An intentionally inert [RadioTransport] that silently discards all operations.
+ *
+ * Used as a safe default when no valid device address is configured or when the requested transport type is
+ * unsupported. All method calls are no-ops β it never connects, never sends data, and never signals lifecycle events to
+ * the service layer.
+ */
+class NopRadioTransport(val address: String) : RadioTransport {
override fun handleSendToRadio(p: ByteArray) {
// No-op
}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/StreamInterface.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/StreamTransport.kt
similarity index 66%
rename from core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/StreamInterface.kt
rename to core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/StreamTransport.kt
index d72c9d0d51..ff2e5e33e5 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/StreamInterface.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/radio/StreamTransport.kt
@@ -17,10 +17,11 @@
package org.meshtastic.core.network.radio
import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CoroutineScope
import org.meshtastic.core.common.util.handledLaunch
import org.meshtastic.core.network.transport.StreamFrameCodec
-import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.core.repository.RadioTransport
+import org.meshtastic.core.repository.RadioTransportCallback
/**
* An interface that assumes we are talking to a meshtastic device over some sort of stream connection (serial or TCP
@@ -28,9 +29,11 @@ import org.meshtastic.core.repository.RadioTransport
*
* Delegates framing logic to [StreamFrameCodec] from `core:network`.
*/
-abstract class StreamInterface(protected val service: RadioInterfaceService) : RadioTransport {
+abstract class StreamTransport(protected val callback: RadioTransportCallback, protected val scope: CoroutineScope) :
+ RadioTransport {
- private val codec = StreamFrameCodec(onPacketReceived = { service.handleFromRadio(it) }, logTag = "StreamInterface")
+ private val codec =
+ StreamFrameCodec(onPacketReceived = { callback.handleFromRadio(it) }, logTag = "StreamTransport")
override fun close() {
Logger.d { "Closing stream for good" }
@@ -38,33 +41,34 @@ abstract class StreamInterface(protected val service: RadioInterfaceService) : R
}
/**
- * Tell MeshService our device has gone away, but wait for it to come back
+ * Notify the transport callback that our device has gone away, but wait for it to come back.
*
- * @param waitForStopped if true we should wait for the manager to finish - must be false if called from inside the
- * manager callbacks
+ * @param waitForStopped if true we should wait for the transport to finish - must be false if called from inside
+ * transport callbacks
* @param isPermanent true if the device is definitely gone (e.g. USB unplugged), false if it may come back (e.g.
- * TCP transient disconnect). Defaults to true for serial β subclasses like [TCPInterface] override with false.
+ * TCP transient disconnect). Defaults to true for serial β subclasses may override with false.
*/
protected open fun onDeviceDisconnect(waitForStopped: Boolean, isPermanent: Boolean = true) {
- service.onDisconnect(isPermanent = isPermanent)
+ callback.onDisconnect(isPermanent = isPermanent)
}
protected open fun connect() {
- // Before telling mesh service, send a few START1s to wake a sleeping device
+ // Before connecting, send a few START1s to wake a sleeping device
sendBytes(StreamFrameCodec.WAKE_BYTES)
// Now tell clients they can (finally use the api)
- service.onConnect()
+ callback.onConnect()
}
+ /** Writes raw bytes to the underlying stream (serial port, TCP socket, etc.). */
abstract fun sendBytes(p: ByteArray)
- // If subclasses need to flush at the end of a packet they can implement
+ /** Flushes buffered bytes to the underlying stream. No-op by default. */
open fun flushBytes() {}
override fun handleSendToRadio(p: ByteArray) {
// This method is called from a continuation and it might show up late, so check for uart being null
- service.serviceScope.handledLaunch { codec.frameAndSend(p, ::sendBytes, ::flushBytes) }
+ scope.handledLaunch { codec.frameAndSend(p, ::sendBytes, ::flushBytes) }
}
/** Process a single incoming byte through the stream framing state machine. */
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
index 6be47c8eb2..5e4ffa91db 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
@@ -18,12 +18,15 @@ package org.meshtastic.core.network.repository
import co.touchlab.kermit.Logger
import io.github.davidepianca98.MQTTClient
+import io.github.davidepianca98.mqtt.MQTTException
import io.github.davidepianca98.mqtt.MQTTVersion
import io.github.davidepianca98.mqtt.Subscription
import io.github.davidepianca98.mqtt.packets.Qos
import io.github.davidepianca98.mqtt.packets.mqttv5.ReasonCode
import io.github.davidepianca98.mqtt.packets.mqttv5.SubscriptionOptions
+import io.github.davidepianca98.socket.IOException
import io.github.davidepianca98.socket.tls.TLSClientSettings
+import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
@@ -36,9 +39,12 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import kotlinx.serialization.ExperimentalSerializationApi
+import kotlinx.serialization.SerializationException
import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonDecodingException
import okio.ByteString.Companion.toByteString
import org.koin.core.annotation.Single
+import org.meshtastic.core.di.CoroutineDispatchers
import org.meshtastic.core.model.MqttJsonPayload
import org.meshtastic.core.model.util.subscribeList
import org.meshtastic.core.repository.NodeRepository
@@ -50,7 +56,7 @@ import kotlin.concurrent.Volatile
class MQTTRepositoryImpl(
private val radioConfigRepository: RadioConfigRepository,
private val nodeRepository: NodeRepository,
- dispatchers: org.meshtastic.core.di.CoroutineDispatchers,
+ dispatchers: CoroutineDispatchers,
) : MQTTRepository {
companion object {
@@ -78,14 +84,15 @@ class MQTTRepositoryImpl(
@Suppress("TooGenericExceptionCaught")
override fun disconnect() {
Logger.i { "MQTT Disconnecting" }
+ val c = client
+ client = null // Null first to prevent re-entrant disconnect
try {
- client?.disconnect(ReasonCode.SUCCESS)
+ c?.disconnect(ReasonCode.SUCCESS)
} catch (e: Exception) {
Logger.w(e) { "MQTT clean disconnect failed" }
}
clientJob?.cancel()
clientJob = null
- client = null
}
@OptIn(ExperimentalUnsignedTypes::class)
@@ -123,10 +130,10 @@ class MQTTRepositoryImpl(
Logger.d { "MQTT parsed JSON payload successfully" }
trySend(MqttClientProxyMessage(topic = topic, text = jsonStr, retained = packet.retain))
- } catch (e: kotlinx.serialization.json.JsonDecodingException) {
+ } catch (e: JsonDecodingException) {
@OptIn(ExperimentalSerializationApi::class)
Logger.e(e) { "Failed to parse MQTT JSON: ${e.shortMessage} (path: ${e.path})" }
- } catch (e: kotlinx.serialization.SerializationException) {
+ } catch (e: SerializationException) {
Logger.e(e) { "Failed to parse MQTT JSON: ${e.message}" }
} catch (e: IllegalArgumentException) {
Logger.e(e) { "Failed to parse MQTT JSON: ${e.message}" }
@@ -180,11 +187,11 @@ class MQTTRepositoryImpl(
// Reset backoff so the next reconnect starts with the minimum delay.
reconnectDelay = INITIAL_RECONNECT_DELAY_MS
Logger.w { "MQTT client loop ended normally, reconnecting in ${reconnectDelay}ms" }
- } catch (e: io.github.davidepianca98.mqtt.MQTTException) {
+ } catch (e: MQTTException) {
Logger.e(e) { "MQTT Client loop error (MQTT), reconnecting in ${reconnectDelay}ms" }
- } catch (e: io.github.davidepianca98.socket.IOException) {
+ } catch (e: IOException) {
Logger.e(e) { "MQTT Client loop error (IO), reconnecting in ${reconnectDelay}ms" }
- } catch (e: kotlinx.coroutines.CancellationException) {
+ } catch (e: CancellationException) {
Logger.i { "MQTT Client loop cancelled" }
throw e
}
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/transport/HeartbeatSender.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/transport/HeartbeatSender.kt
new file mode 100644
index 0000000000..045d3b7ec8
--- /dev/null
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/transport/HeartbeatSender.kt
@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.network.transport
+
+import co.touchlab.kermit.Logger
+import org.meshtastic.proto.Heartbeat
+import org.meshtastic.proto.ToRadio
+import kotlin.concurrent.atomics.AtomicInt
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
+
+/**
+ * Shared heartbeat sender for Meshtastic radio transports.
+ *
+ * Constructs and sends a `ToRadio(heartbeat = Heartbeat(nonce = ...))` message to keep the firmware's idle timer from
+ * expiring. Each call uses a monotonically increasing nonce to prevent the firmware's per-connection duplicate-write
+ * filter from silently dropping it.
+ *
+ * @param sendToRadio callback to transmit the encoded heartbeat bytes to the radio
+ * @param afterHeartbeat optional suspend callback invoked after sending (e.g. to schedule a drain)
+ * @param logTag tag for log messages
+ */
+class HeartbeatSender(
+ private val sendToRadio: (ByteArray) -> Unit,
+ private val afterHeartbeat: (suspend () -> Unit)? = null,
+ private val logTag: String = "HeartbeatSender",
+) {
+ @OptIn(ExperimentalAtomicApi::class)
+ private val nonce = AtomicInt(0)
+
+ /**
+ * Sends a heartbeat to the radio.
+ *
+ * The firmware responds to heartbeats by queuing a `queueStatus` FromRadio packet, proving the link is alive and
+ * keeping the local node's lastHeard timestamp current.
+ */
+ @OptIn(ExperimentalAtomicApi::class)
+ suspend fun sendHeartbeat() {
+ val n = nonce.fetchAndAdd(1)
+ Logger.v { "[$logTag] Sending ToRadio heartbeat (nonce=$n)" }
+ sendToRadio(ToRadio(heartbeat = Heartbeat(nonce = n)).encode())
+ afterHeartbeat?.invoke()
+ }
+}
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioInterfaceTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportTest.kt
similarity index 70%
rename from core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioInterfaceTest.kt
rename to core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportTest.kt
index d4a41ba953..f1049f897d 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioInterfaceTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleRadioTransportTest.kt
@@ -36,10 +36,9 @@ import org.meshtastic.core.testing.FakeBluetoothRepository
import kotlin.test.BeforeTest
import kotlin.test.Test
import kotlin.test.assertEquals
-import kotlin.time.Duration.Companion.seconds
@OptIn(ExperimentalCoroutinesApi::class)
-class BleRadioInterfaceTest {
+class BleRadioTransportTest {
private val testScope = TestScope()
private val scanner = FakeBleScanner()
@@ -56,66 +55,69 @@ class BleRadioInterfaceTest {
}
@Test
- fun `connect attempts to scan and connect via init`() = runTest {
+ fun `connect attempts to scan and connect via start`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Device")
scanner.emitDevice(device)
- val bleInterface =
- BleRadioInterface(
- serviceScope = testScope,
+ val bleTransport =
+ BleRadioTransport(
+ scope = testScope,
scanner = scanner,
bluetoothRepository = bluetoothRepository,
connectionFactory = connectionFactory,
- service = service,
+ callback = service,
address = address,
)
+ bleTransport.start()
- // init starts connect() which is async
+ // start() begins connect() which is async
// In a real test we'd verify the connection state,
// but for now this confirms it works with the fakes.
- assertEquals(address, bleInterface.address)
+ assertEquals(address, bleTransport.address)
}
@Test
fun `address returns correct value`() {
- val bleInterface =
- BleRadioInterface(
- serviceScope = testScope,
+ val bleTransport =
+ BleRadioTransport(
+ scope = testScope,
scanner = scanner,
bluetoothRepository = bluetoothRepository,
connectionFactory = connectionFactory,
- service = service,
+ callback = service,
address = address,
)
- assertEquals(address, bleInterface.address)
+ assertEquals(address, bleTransport.address)
}
/**
- * After [RECONNECT_FAILURE_THRESHOLD] consecutive connection failures, [RadioInterfaceService.onDisconnect] must be
- * called so the higher layers can react (e.g. start the device-sleep timeout in [MeshConnectionManagerImpl]).
+ * After [BleReconnectPolicy.DEFAULT_FAILURE_THRESHOLD] consecutive connection failures,
+ * [RadioInterfaceService.onDisconnect] must be called so the higher layers can react (e.g. start the device-sleep
+ * timeout in [MeshConnectionManagerImpl]).
*
- * Virtual-time breakdown (RECONNECT_FAILURE_THRESHOLD = 3): t = 1 000 ms β iteration 1 settle delay elapses,
+ * Virtual-time breakdown (DEFAULT_FAILURE_THRESHOLD = 3): t = 1 000 ms β iteration 1 settle delay elapses,
* connectAndAwait throws, backoff 5 s starts t = 6 000 ms β backoff ends t = 7 000 ms β iteration 2 settle delay
* elapses, connectAndAwait throws, backoff 10 s starts t = 17 000 ms β backoff ends t = 18 000 ms β iteration 3
* settle delay elapses, connectAndAwait throws β onDisconnect called
*/
@Test
- fun `onDisconnect is called after RECONNECT_FAILURE_THRESHOLD consecutive failures`() = runTest {
+ fun `onDisconnect is called after DEFAULT_FAILURE_THRESHOLD consecutive failures`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Device")
bluetoothRepository.bond(device) // skip BLE scan β device is already bonded
// Make every connectAndAwait call throw so each iteration counts as one failure.
connection.connectException = RadioNotConnectedException("simulated failure")
- val bleInterface =
- BleRadioInterface(
- serviceScope = this,
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
scanner = scanner,
bluetoothRepository = bluetoothRepository,
connectionFactory = connectionFactory,
- service = service,
+ callback = service,
address = address,
)
+ bleTransport.start()
// Advance through exactly 3 failure iterations (β18 001 ms virtual time).
// The 4th iteration's backoff hasn't elapsed yet, so the coroutine is suspended
@@ -125,12 +127,12 @@ class BleRadioInterfaceTest {
verify { service.onDisconnect(any(), any()) }
// Cancel the reconnect loop so runTest can complete.
- bleInterface.close()
+ bleTransport.close()
}
/**
- * After [RECONNECT_MAX_FAILURES] (10) consecutive failures, the reconnect loop should stop and signal a permanent
- * disconnect. This prevents infinite battery drain when the device is genuinely offline.
+ * After [BleReconnectPolicy.DEFAULT_MAX_FAILURES] (10) consecutive failures, the reconnect loop should stop and
+ * signal a permanent disconnect. This prevents infinite battery drain when the device is genuinely offline.
*
* Time budget for 10 failures with bonded device (no scan): Each iteration = 1s settle + connectAndAwait throw +
* backoff Backoffs: 5s, 10s, 20s, 40s, 60s, 60s, 60s, 60s, 60s, (exit at failure 10 before backoff) Total β 10Γ1s
@@ -138,22 +140,23 @@ class BleRadioInterfaceTest {
* variance.
*/
@Test
- fun `reconnect loop stops after RECONNECT_MAX_FAILURES with permanent disconnect`() = runTest {
+ fun `reconnect loop stops after DEFAULT_MAX_FAILURES with permanent disconnect`() = runTest {
val device = FakeBleDevice(address = address, name = "Test Device")
bluetoothRepository.bond(device)
connection.connectException = RadioNotConnectedException("simulated failure")
every { service.onDisconnect(any(), any()) } returns Unit
- val bleInterface =
- BleRadioInterface(
- serviceScope = this,
+ val bleTransport =
+ BleRadioTransport(
+ scope = this,
scanner = scanner,
bluetoothRepository = bluetoothRepository,
connectionFactory = connectionFactory,
- service = service,
+ callback = service,
address = address,
)
+ bleTransport.start()
// Advance enough time for all 10 failures to occur.
advanceTimeBy(400_001L)
@@ -161,18 +164,6 @@ class BleRadioInterfaceTest {
// Should have been called with isPermanent=true at least once (the final call).
verify { service.onDisconnect(isPermanent = true, errorMessage = any()) }
- bleInterface.close()
- }
-
- @Test
- fun `computeReconnectBackoff returns correct backoff values`() {
- assertEquals(5.seconds, computeReconnectBackoff(0))
- assertEquals(5.seconds, computeReconnectBackoff(1))
- assertEquals(10.seconds, computeReconnectBackoff(2))
- assertEquals(20.seconds, computeReconnectBackoff(3))
- assertEquals(40.seconds, computeReconnectBackoff(4))
- assertEquals(60.seconds, computeReconnectBackoff(5))
- assertEquals(60.seconds, computeReconnectBackoff(10))
- assertEquals(60.seconds, computeReconnectBackoff(100))
+ bleTransport.close()
}
}
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleReconnectPolicyTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleReconnectPolicyTest.kt
new file mode 100644
index 0000000000..a6a7aa82cb
--- /dev/null
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/BleReconnectPolicyTest.kt
@@ -0,0 +1,277 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.network.radio
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+
+class BleReconnectPolicyTest {
+
+ @Test
+ fun `stable disconnect resets failures and returns Continue`() {
+ val policy = BleReconnectPolicy()
+ // Simulate one prior failure
+ policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+ assertEquals(1, policy.consecutiveFailures)
+
+ // Now a stable disconnect should reset
+ val action =
+ policy.processOutcome(BleReconnectPolicy.Outcome.Disconnected(wasStable = true, wasIntentional = false))
+ assertEquals(BleReconnectPolicy.Action.Continue, action)
+ assertEquals(0, policy.consecutiveFailures)
+ }
+
+ @Test
+ fun `intentional disconnect resets failures and returns Continue`() {
+ val policy = BleReconnectPolicy()
+ policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+
+ val action =
+ policy.processOutcome(BleReconnectPolicy.Outcome.Disconnected(wasStable = false, wasIntentional = true))
+ assertEquals(BleReconnectPolicy.Action.Continue, action)
+ assertEquals(0, policy.consecutiveFailures)
+ }
+
+ @Test
+ fun `unstable disconnect increments failures`() {
+ val policy = BleReconnectPolicy()
+ val action =
+ policy.processOutcome(BleReconnectPolicy.Outcome.Disconnected(wasStable = false, wasIntentional = false))
+ assertEquals(1, policy.consecutiveFailures)
+ assertTrue(action is BleReconnectPolicy.Action.Retry)
+ }
+
+ @Test
+ fun `failure at threshold signals transient disconnect`() {
+ val policy = BleReconnectPolicy(failureThreshold = 3)
+ // Accumulate failures up to threshold
+ repeat(2) { policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test"))) }
+ val action = policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+ assertEquals(3, policy.consecutiveFailures)
+ assertTrue(action is BleReconnectPolicy.Action.SignalTransient)
+ }
+
+ @Test
+ fun `failure at max gives up permanently`() {
+ val policy = BleReconnectPolicy(maxFailures = 3)
+ repeat(2) { policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test"))) }
+ val action = policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+ assertEquals(BleReconnectPolicy.Action.GiveUp, action)
+ }
+
+ @Test
+ fun `backoff increases with consecutive failures`() {
+ val policy = BleReconnectPolicy()
+ val backoffs =
+ (1..5).map { i ->
+ val action = policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+ when (action) {
+ is BleReconnectPolicy.Action.Retry -> action.backoff
+ is BleReconnectPolicy.Action.SignalTransient -> action.backoff
+ else -> error("Unexpected action: $action")
+ }
+ }
+ // Verify backoffs are non-decreasing
+ for (i in 0 until backoffs.size - 1) {
+ assertTrue(backoffs[i] <= backoffs[i + 1], "Expected ${backoffs[i]} <= ${backoffs[i + 1]}")
+ }
+ }
+
+ @Test
+ fun `custom backoff strategy is used`() {
+ val customBackoff = 42.seconds
+ val policy = BleReconnectPolicy(backoffStrategy = { customBackoff })
+ val action = policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+ assertTrue(action is BleReconnectPolicy.Action.Retry)
+ assertEquals(customBackoff, action.backoff)
+ }
+
+ @Test
+ fun `maxFailures equal to failureThreshold gives up without signalling transient`() {
+ val policy = BleReconnectPolicy(maxFailures = 3, failureThreshold = 3)
+ repeat(2) { policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test"))) }
+ val action = policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+ // GiveUp takes priority over SignalTransient when both thresholds are the same
+ assertEquals(BleReconnectPolicy.Action.GiveUp, action)
+ }
+
+ @Test
+ fun `failure count resets after stable disconnect then re-increments`() {
+ val policy = BleReconnectPolicy()
+ // Accumulate two failures
+ repeat(2) { policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test"))) }
+ assertEquals(2, policy.consecutiveFailures)
+
+ // Stable disconnect resets
+ policy.processOutcome(BleReconnectPolicy.Outcome.Disconnected(wasStable = true, wasIntentional = false))
+ assertEquals(0, policy.consecutiveFailures)
+
+ // New failure starts from 1
+ policy.processOutcome(BleReconnectPolicy.Outcome.Failed(RuntimeException("test")))
+ assertEquals(1, policy.consecutiveFailures)
+ }
+
+ // region execute() loop tests
+
+ @OptIn(ExperimentalCoroutinesApi::class)
+ @Test
+ fun `execute gives up after maxFailures and calls onPermanentDisconnect`() = runTest {
+ val policy =
+ BleReconnectPolicy(maxFailures = 3, settleDelay = 1.milliseconds, backoffStrategy = { 1.milliseconds })
+ var permanentError: Throwable? = null
+ var permanentCalled = false
+ var transientCalled = false
+
+ policy.execute(
+ attempt = { BleReconnectPolicy.Outcome.Failed(RuntimeException("connection failed")) },
+ onTransientDisconnect = { transientCalled = true },
+ onPermanentDisconnect = { error ->
+ permanentCalled = true
+ permanentError = error
+ },
+ )
+
+ assertTrue(permanentCalled, "onPermanentDisconnect should have been called")
+ assertNotNull(permanentError, "error should be passed to onPermanentDisconnect")
+ assertEquals("connection failed", permanentError?.message)
+ assertEquals(3, policy.consecutiveFailures)
+ // failureThreshold defaults to 3, same as maxFailures here, so GiveUp takes priority
+ assertTrue(!transientCalled, "onTransientDisconnect should not be called when GiveUp fires first")
+ }
+
+ @OptIn(ExperimentalCoroutinesApi::class)
+ @Test
+ fun `execute calls onTransientDisconnect at threshold then continues retrying`() = runTest {
+ var attemptCount = 0
+ val policy =
+ BleReconnectPolicy(
+ maxFailures = 5,
+ failureThreshold = 2,
+ settleDelay = 1.milliseconds,
+ backoffStrategy = { 1.milliseconds },
+ )
+ var transientCount = 0
+
+ policy.execute(
+ attempt = {
+ attemptCount++
+ BleReconnectPolicy.Outcome.Failed(RuntimeException("fail #$attemptCount"))
+ },
+ onTransientDisconnect = { transientCount++ },
+ onPermanentDisconnect = {},
+ )
+
+ assertEquals(5, attemptCount, "should attempt exactly maxFailures times")
+ // Transient is signalled for failures 2, 3, 4 (at or above threshold, below maxFailures)
+ assertEquals(3, transientCount, "should signal transient for each failure at or above threshold")
+ }
+
+ @OptIn(ExperimentalCoroutinesApi::class)
+ @Test
+ fun `execute continues immediately after stable disconnect`() = runTest {
+ var attemptCount = 0
+ val policy =
+ BleReconnectPolicy(maxFailures = 5, settleDelay = 1.milliseconds, backoffStrategy = { 1.milliseconds })
+
+ policy.execute(
+ attempt = {
+ attemptCount++
+ if (attemptCount <= 2) {
+ // First two attempts connect briefly and disconnect stably
+ BleReconnectPolicy.Outcome.Disconnected(wasStable = true, wasIntentional = false)
+ } else {
+ // Then fail until maxFailures
+ BleReconnectPolicy.Outcome.Failed(RuntimeException("fail"))
+ }
+ },
+ onTransientDisconnect = {},
+ onPermanentDisconnect = {},
+ )
+
+ // 2 stable disconnects + 5 failures (counter resets after each stable, so needs 5 more to hit max)
+ assertEquals(7, attemptCount)
+ assertEquals(5, policy.consecutiveFailures)
+ }
+
+ @OptIn(ExperimentalCoroutinesApi::class)
+ @Test
+ fun `execute passes null error for unstable disconnect at threshold`() = runTest {
+ val policy =
+ BleReconnectPolicy(
+ maxFailures = 5,
+ failureThreshold = 2,
+ settleDelay = 1.milliseconds,
+ backoffStrategy = { 1.milliseconds },
+ )
+ val transientErrors = mutableListOf<Throwable?>()
+ var attemptCount = 0
+
+ policy.execute(
+ attempt = {
+ attemptCount++
+ // Use unstable disconnects (not Failed) so lastError is null
+ BleReconnectPolicy.Outcome.Disconnected(wasStable = false, wasIntentional = false)
+ },
+ onTransientDisconnect = { error -> transientErrors.add(error) },
+ onPermanentDisconnect = {},
+ )
+
+ // Disconnected outcomes don't have errors, so all transient callbacks get null
+ assertTrue(transientErrors.all { it == null }, "Disconnected outcomes should pass null error")
+ }
+
+ @OptIn(ExperimentalCoroutinesApi::class)
+ @Test
+ fun `execute stops when coroutine is cancelled`() = runTest {
+ var attemptCount = 0
+ val policy =
+ BleReconnectPolicy(maxFailures = 100, settleDelay = 1.milliseconds, backoffStrategy = { 1.milliseconds })
+
+ val job =
+ backgroundScope.launch {
+ policy.execute(
+ attempt = {
+ attemptCount++
+ // Always succeed stably β loop should run until cancelled
+ BleReconnectPolicy.Outcome.Disconnected(wasStable = true, wasIntentional = false)
+ },
+ onTransientDisconnect = {},
+ onPermanentDisconnect = {},
+ )
+ }
+
+ // Let a few iterations run, then cancel
+ advanceTimeBy(50)
+ job.cancel()
+ advanceUntilIdle()
+
+ // Should have made some attempts but not reached maxFailures
+ assertTrue(attemptCount > 0, "should have attempted at least once")
+ assertTrue(attemptCount < 100, "should not have exhausted all failures β was cancelled")
+ }
+
+ // endregion
+}
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/ReconnectBackoffTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/ReconnectBackoffTest.kt
index c4e64d36a4..f3514c752a 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/ReconnectBackoffTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/ReconnectBackoffTest.kt
@@ -22,7 +22,7 @@ import kotlin.test.assertTrue
import kotlin.time.Duration.Companion.seconds
/**
- * Tests the exponential backoff schedule used by [BleRadioInterface] when consecutive connection attempts fail. The
+ * Tests the exponential backoff schedule used by [BleRadioTransport] when consecutive connection attempts fail. The
* schedule is: failure #1 β 5 s failure #2 β 10 s failure #3 β 20 s failure #4 β 40 s failure #5+ β 60 s (capped)
*/
class ReconnectBackoffTest {
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/StreamInterfaceTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/StreamTransportTest.kt
similarity index 75%
rename from core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/StreamInterfaceTest.kt
rename to core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/StreamTransportTest.kt
index 4c4e9b4be4..6faa692173 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/StreamInterfaceTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/radio/StreamTransportTest.kt
@@ -17,8 +17,6 @@
package org.meshtastic.core.network.radio
import dev.mokkery.MockMode
-import dev.mokkery.answering.returns
-import dev.mokkery.every
import dev.mokkery.mock
import dev.mokkery.verify
import io.kotest.property.Arb
@@ -29,17 +27,16 @@ import io.kotest.property.checkAll
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.runTest
import org.meshtastic.core.network.transport.StreamFrameCodec
-import org.meshtastic.core.repository.RadioInterfaceService
-import kotlin.test.BeforeTest
+import org.meshtastic.core.repository.RadioTransportCallback
import kotlin.test.Test
import kotlin.test.assertTrue
-class StreamInterfaceTest {
+class StreamTransportTest {
- private val radioService: RadioInterfaceService = mock(MockMode.autofill)
- private lateinit var fakeStream: FakeStreamInterface
+ private val callback: RadioTransportCallback = mock(MockMode.autofill)
+ private lateinit var fakeStream: FakeStreamTransport
- class FakeStreamInterface(service: RadioInterfaceService) : StreamInterface(service) {
+ class FakeStreamTransport(callback: RadioTransportCallback, scope: TestScope) : StreamTransport(callback, scope) {
val sentBytes = mutableListOf<ByteArray>()
override fun sendBytes(p: ByteArray) {
@@ -59,21 +56,18 @@ class StreamInterfaceTest {
public override fun connect() = super.connect()
}
- @BeforeTest
- fun setUp() {
- every { radioService.serviceScope } returns TestScope()
- }
+ private val testScope = TestScope()
@Test
fun `handleSendToRadio property test`() = runTest {
- fakeStream = FakeStreamInterface(radioService)
+ fakeStream = FakeStreamTransport(callback, testScope)
checkAll(Arb.byteArray(Arb.int(0, 512), Arb.byte())) { payload -> fakeStream.handleSendToRadio(payload) }
}
@Test
fun `readChar property test`() = runTest {
- fakeStream = FakeStreamInterface(radioService)
+ fakeStream = FakeStreamTransport(callback, testScope)
checkAll(Arb.byteArray(Arb.int(0, 100), Arb.byte())) { data ->
data.forEach { fakeStream.feed(it) }
@@ -83,11 +77,11 @@ class StreamInterfaceTest {
@Test
fun `connect sends wake bytes`() {
- fakeStream = FakeStreamInterface(radioService)
+ fakeStream = FakeStreamTransport(callback, testScope)
fakeStream.connect()
assertTrue(fakeStream.sentBytes.isNotEmpty())
assertTrue(fakeStream.sentBytes[0].contentEquals(StreamFrameCodec.WAKE_BYTES))
- verify { radioService.onConnect() }
+ verify { callback.onConnect() }
}
}
diff --git a/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/radio/TCPInterface.kt b/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/radio/TcpRadioTransport.kt
similarity index 57%
rename from core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/radio/TCPInterface.kt
rename to core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/radio/TcpRadioTransport.kt
index 0ffb731cf0..7b1106dc46 100644
--- a/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/radio/TCPInterface.kt
+++ b/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/radio/TcpRadioTransport.kt
@@ -17,76 +17,76 @@
package org.meshtastic.core.network.radio
import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CoroutineScope
import org.meshtastic.core.common.util.handledLaunch
import org.meshtastic.core.di.CoroutineDispatchers
import org.meshtastic.core.network.transport.StreamFrameCodec
import org.meshtastic.core.network.transport.TcpTransport
-import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.core.repository.RadioTransport
+import org.meshtastic.core.repository.RadioTransportCallback
+import kotlin.concurrent.Volatile
/**
- * Android TCP radio interface β thin adapter over the shared [TcpTransport] from `core:network`.
+ * TCP radio transport β thin adapter over the shared [TcpTransport] from `core:network`.
*
- * Manages the mapping between the Android-specific [StreamInterface]/[RadioTransport] contract and the shared transport
- * layer.
+ * Implements [RadioTransport] directly via composition over [TcpTransport], delegating send/receive to the transport
+ * and calling [RadioTransportCallback] for lifecycle events. This avoids the previous inheritance from
+ * [StreamTransport] which created a dead [StreamFrameCodec] and required overriding `sendBytes` as a no-op.
*/
-open class TCPInterface(
- service: RadioInterfaceService,
+open class TcpRadioTransport(
+ private val callback: RadioTransportCallback,
+ private val scope: CoroutineScope,
private val dispatchers: CoroutineDispatchers,
private val address: String,
-) : StreamInterface(service) {
+) : RadioTransport {
companion object {
const val SERVICE_PORT = StreamFrameCodec.DEFAULT_TCP_PORT
}
+ /** Guards against a double [RadioTransportCallback.onDisconnect] when [close] triggers [TcpTransport.stop]. */
+ @Volatile private var closing = false
+
private val transport =
TcpTransport(
dispatchers = dispatchers,
- scope = service.serviceScope,
+ scope = scope,
listener =
object : TcpTransport.Listener {
override fun onConnected() {
- super@TCPInterface.connect()
+ callback.onConnect()
}
override fun onDisconnected() {
- // Transport already performed teardown; only propagate lifecycle to StreamInterface.
+ if (closing) return // close() will fire the permanent disconnect itself
// TCP disconnects are transient (not permanent) β the transport will auto-reconnect.
- super@TCPInterface.onDeviceDisconnect(false, isPermanent = false)
+ callback.onDisconnect(isPermanent = false)
}
override fun onPacketReceived(bytes: ByteArray) {
- service.handleFromRadio(bytes)
+ callback.handleFromRadio(bytes)
}
},
- logTag = "TCPInterface[$address]",
+ logTag = "TcpRadioTransport[$address]",
)
- init {
- connect()
- }
-
- override fun sendBytes(p: ByteArray) {
- // Direct byte sending is handled by the transport; this is used by StreamInterface for serial compat
- Logger.d { "[$address] TCPInterface.sendBytes delegated to transport" }
+ override fun start() {
+ transport.start(address)
}
- override fun onDeviceDisconnect(waitForStopped: Boolean, isPermanent: Boolean) {
+ override fun close() {
+ Logger.d { "[$address] Closing TCP transport" }
+ closing = true
transport.stop()
- super.onDeviceDisconnect(waitForStopped, isPermanent = false)
- }
-
- override fun connect() {
- transport.start(address)
+ callback.onDisconnect(isPermanent = true)
}
override fun keepAlive() {
Logger.d { "[$address] TCP keepAlive" }
- service.serviceScope.handledLaunch { transport.sendHeartbeat() }
+ scope.handledLaunch { transport.sendHeartbeat() }
}
override fun handleSendToRadio(p: ByteArray) {
- service.serviceScope.handledLaunch { transport.sendPacket(p) }
+ scope.handledLaunch { transport.sendPacket(p) }
}
}
diff --git a/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/transport/TcpTransport.kt b/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/transport/TcpTransport.kt
index 264e42f89e..172423470d 100644
--- a/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/transport/TcpTransport.kt
+++ b/core/network/src/jvmAndroidMain/kotlin/org/meshtastic/core/network/transport/TcpTransport.kt
@@ -24,7 +24,6 @@ import kotlinx.coroutines.withContext
import org.meshtastic.core.common.util.handledLaunch
import org.meshtastic.core.common.util.nowMillis
import org.meshtastic.core.di.CoroutineDispatchers
-import org.meshtastic.proto.Heartbeat
import org.meshtastic.proto.ToRadio
import java.io.BufferedInputStream
import java.io.BufferedOutputStream
@@ -34,13 +33,14 @@ import java.net.InetAddress
import java.net.Socket
import java.net.SocketTimeoutException
import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
/**
* Shared JVM TCP transport for Meshtastic radios.
*
* Manages the TCP socket lifecycle (connect, read loop, reconnect with backoff) and uses [StreamFrameCodec] for the
- * START1/START2 stream framing protocol. Heartbeat scheduling is owned by [SharedRadioInterfaceService]; this class
- * only exposes [sendHeartbeat] for external callers.
+ * START1/START2 stream framing protocol. [sendHeartbeat] sends a heartbeat with a monotonically-increasing nonce so the
+ * firmware's per-connection duplicate-write filter does not silently drop it.
*
* Used by Android and Desktop via the shared `SharedRadioInterfaceService`.
*/
@@ -109,6 +109,8 @@ class TcpTransport(
@Volatile private var timeoutEvents: Int = 0
+ private val heartbeatNonce = AtomicInteger(0)
+
/** Whether the transport is currently connected. */
val isConnected: Boolean
get() {
@@ -146,9 +148,10 @@ class TcpTransport(
bytesSent += payload.size
}
- /** Send a heartbeat packet to keep the connection alive. */
+ /** Send a heartbeat packet with a monotonically-increasing nonce to keep the connection alive. */
suspend fun sendHeartbeat() {
- val heartbeat = ToRadio(heartbeat = Heartbeat())
+ val nonce = heartbeatNonce.getAndIncrement()
+ val heartbeat = ToRadio(heartbeat = org.meshtastic.proto.Heartbeat(nonce = nonce))
sendPacket(heartbeat.encode())
}
diff --git a/core/network/src/jvmMain/kotlin/org/meshtastic/core/network/SerialTransport.kt b/core/network/src/jvmMain/kotlin/org/meshtastic/core/network/SerialTransport.kt
index a773312673..d43063d528 100644
--- a/core/network/src/jvmMain/kotlin/org/meshtastic/core/network/SerialTransport.kt
+++ b/core/network/src/jvmMain/kotlin/org/meshtastic/core/network/SerialTransport.kt
@@ -19,18 +19,19 @@ package org.meshtastic.core.network
import co.touchlab.kermit.Logger
import com.fazecast.jSerialComm.SerialPort
import com.fazecast.jSerialComm.SerialPortTimeoutException
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import org.meshtastic.core.di.CoroutineDispatchers
-import org.meshtastic.core.network.radio.StreamInterface
-import org.meshtastic.core.repository.RadioInterfaceService
-import org.meshtastic.proto.Heartbeat
-import org.meshtastic.proto.ToRadio
+import org.meshtastic.core.network.radio.StreamTransport
+import org.meshtastic.core.network.transport.HeartbeatSender
+import org.meshtastic.core.repository.RadioTransportCallback
import java.io.File
/**
- * JVM-specific implementation of [RadioTransport] using jSerialComm. Uses [StreamInterface] for START1/START2 packet
+ * JVM-specific implementation of [RadioTransport] using jSerialComm. Uses [StreamTransport] for START1/START2 packet
* framing.
*
* Use the [open] factory method instead of the constructor directly to ensure the serial port is opened and the read
@@ -40,12 +41,15 @@ class SerialTransport
private constructor(
private val portName: String,
private val baudRate: Int = DEFAULT_BAUD_RATE,
- service: RadioInterfaceService,
+ callback: RadioTransportCallback,
+ scope: CoroutineScope,
private val dispatchers: CoroutineDispatchers,
-) : StreamInterface(service) {
+) : StreamTransport(callback, scope) {
private var serialPort: SerialPort? = null
private var readJob: Job? = null
+ private val heartbeatSender = HeartbeatSender(sendToRadio = ::handleSendToRadio, logTag = "Serial[$portName]")
+
/** Attempts to open the serial port and starts the read loop. Returns true if successful, false otherwise. */
private fun startConnection(): Boolean {
return try {
@@ -57,7 +61,7 @@ private constructor(
port.setDTR()
port.setRTS()
Logger.i { "[$portName] Serial port opened (baud=$baudRate)" }
- super.connect() // Sends WAKE_BYTES and signals service.onConnect()
+ super.connect() // Sends WAKE_BYTES and signals callback.onConnect()
startReadLoop(port)
true
} else {
@@ -74,7 +78,7 @@ private constructor(
private fun startReadLoop(port: SerialPort) {
Logger.d { "[$portName] Starting serial read loop" }
readJob =
- service.serviceScope.launch(dispatchers.io) {
+ scope.launch(dispatchers.io) {
val input = port.inputStream
val buffer = ByteArray(READ_BUFFER_SIZE)
try {
@@ -91,7 +95,7 @@ private constructor(
}
} catch (_: SerialPortTimeoutException) {
// Expected timeout when no data is available
- } catch (e: kotlinx.coroutines.CancellationException) {
+ } catch (e: CancellationException) {
throw e
} catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
if (isActive) {
@@ -102,7 +106,7 @@ private constructor(
reading = false
}
}
- } catch (e: kotlinx.coroutines.CancellationException) {
+ } catch (e: CancellationException) {
throw e
} catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
if (isActive) {
@@ -140,11 +144,9 @@ private constructor(
}
override fun keepAlive() {
- // Send a ToRadio heartbeat so the firmware resets its idle timer and responds with
- // a FromRadio queueStatus β proving the serial link is alive. Without this, the
- // serial transport has no way to detect a silently dead device.
- Logger.d { "[$portName] Serial keepAlive β sending heartbeat" }
- handleSendToRadio(ToRadio(heartbeat = Heartbeat()).encode())
+ // Delegate to HeartbeatSender which sends a ToRadio heartbeat to prove the
+ // serial link is alive.
+ scope.launch { heartbeatSender.sendHeartbeat() }
}
private fun closePortResources() {
@@ -168,19 +170,20 @@ private constructor(
/**
* Creates and opens a [SerialTransport]. If the port cannot be opened, the transport signals a permanent
- * disconnect to the [service] and returns the (non-connected) instance.
+ * disconnect to the [callback] and returns the (non-connected) instance.
*/
fun open(
portName: String,
baudRate: Int = DEFAULT_BAUD_RATE,
- service: RadioInterfaceService,
+ callback: RadioTransportCallback,
+ scope: CoroutineScope,
dispatchers: CoroutineDispatchers,
): SerialTransport {
- val transport = SerialTransport(portName, baudRate, service, dispatchers)
+ val transport = SerialTransport(portName, baudRate, callback, scope, dispatchers)
if (!transport.startConnection()) {
val errorMessage = diagnoseOpenFailure(portName)
Logger.w { "[$portName] Serial port could not be opened; signalling disconnect. $errorMessage" }
- service.onDisconnect(isPermanent = true, errorMessage = errorMessage)
+ callback.onDisconnect(isPermanent = true, errorMessage = errorMessage)
}
return transport
}
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/MeshServiceNotifications.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/MeshServiceNotifications.kt
index 30aade8669..a681579434 100644
--- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/MeshServiceNotifications.kt
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/MeshServiceNotifications.kt
@@ -16,6 +16,7 @@
*/
package org.meshtastic.core.repository
+import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.Node
import org.meshtastic.proto.ClientNotification
import org.meshtastic.proto.Telemetry
@@ -28,7 +29,7 @@ interface MeshServiceNotifications {
fun initChannels()
- fun updateServiceStateNotification(state: org.meshtastic.core.model.ConnectionState, telemetry: Telemetry?)
+ fun updateServiceStateNotification(state: ConnectionState, telemetry: Telemetry?)
suspend fun updateMessageNotification(
contactKey: String,
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioInterfaceService.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioInterfaceService.kt
index bb9cea52d2..8dcc21c71c 100644
--- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioInterfaceService.kt
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioInterfaceService.kt
@@ -39,7 +39,7 @@ import org.meshtastic.core.model.MeshActivity
*
* @see ServiceRepository.connectionState
*/
-interface RadioInterfaceService {
+interface RadioInterfaceService : RadioTransportCallback {
/** The device types supported by this platform's radio interface. */
val supportedDeviceTypes: List<DeviceType>
@@ -65,8 +65,8 @@ interface RadioInterfaceService {
/** Flow of the current device address. */
val currentDeviceAddressFlow: StateFlow<String?>
- /** Whether we are currently using a mock interface. */
- fun isMockInterface(): Boolean
+ /** Whether we are currently using a mock transport. */
+ fun isMockTransport(): Boolean
/** Flow of raw data received from the radio. */
val receivedData: SharedFlow<ByteArray>
@@ -89,15 +89,6 @@ interface RadioInterfaceService {
/** Constructs a full radio address for the specific interface type. */
fun toInterfaceAddress(interfaceId: InterfaceId, rest: String): String
- /** Called by an interface when it has successfully connected. */
- fun onConnect()
-
- /** Called by an interface when it has disconnected. */
- fun onDisconnect(isPermanent: Boolean, errorMessage: String? = null)
-
- /** Called by an interface when it has received raw data from the radio. */
- fun handleFromRadio(bytes: ByteArray)
-
/** Flow of user-facing connection error messages (e.g. permission failures). */
val connectionError: SharedFlow<String>
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransport.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransport.kt
index 41015381f9..c6132a103b 100644
--- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransport.kt
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransport.kt
@@ -26,6 +26,14 @@ interface RadioTransport : Closeable {
/** Sends a raw byte array to the radio hardware. */
fun handleSendToRadio(p: ByteArray)
+ /**
+ * Initializes the transport after construction. Called by the factory once the transport has been fully created.
+ *
+ * This separates construction from side effects (connecting, launching coroutines), making transports easier to
+ * test and reason about.
+ */
+ fun start() {}
+
/**
* If we think we are connected, but we don't hear anything from the device, we might be in a zombie state. This
* function can be implemented by transports to see if we are really connected.
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransportCallback.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransportCallback.kt
new file mode 100644
index 0000000000..9771062a5f
--- /dev/null
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransportCallback.kt
@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.repository
+
+/**
+ * Narrow callback interface for transport β service communication.
+ *
+ * Transport implementations ([RadioTransport]) need only these three methods to report lifecycle events and deliver
+ * data. This replaces the previous pattern of passing the full [RadioInterfaceService] to transport constructors,
+ * decoupling transports from the service layer.
+ */
+interface RadioTransportCallback {
+ /** Called when the transport has successfully established a connection. */
+ fun onConnect()
+
+ /**
+ * Called when the transport has disconnected.
+ *
+ * @param isPermanent true if the device is definitely gone (e.g. USB unplugged, max retries exhausted), false if it
+ * may come back (e.g. BLE range, TCP transient).
+ * @param errorMessage optional user-facing error message describing the disconnect reason.
+ */
+ fun onDisconnect(isPermanent: Boolean, errorMessage: String? = null)
+
+ /** Called when the transport has received raw data from the radio. */
+ fun handleFromRadio(bytes: ByteArray)
+}
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransportFactory.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransportFactory.kt
index 918657e999..c3d2abff12 100644
--- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransportFactory.kt
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioTransportFactory.kt
@@ -28,8 +28,8 @@ interface RadioTransportFactory {
/** The device types supported by this factory. */
val supportedDeviceTypes: List<DeviceType>
- /** Whether we are currently forced into using a mock interface (e.g., Firebase Test Lab). */
- fun isMockInterface(): Boolean
+ /** Whether we are currently forced into using a mock transport (e.g., Firebase Test Lab). */
+ fun isMockTransport(): Boolean
/** Creates a transport for the given [address], or a NOP implementation if invalid/unsupported. */
fun createTransport(address: String, service: RadioInterfaceService): RadioTransport
diff --git a/core/service/src/androidMain/kotlin/org/meshtastic/core/service/AndroidRadioControllerImpl.kt b/core/service/src/androidMain/kotlin/org/meshtastic/core/service/AndroidRadioControllerImpl.kt
index a96b3ffc17..af7cb85c20 100644
--- a/core/service/src/androidMain/kotlin/org/meshtastic/core/service/AndroidRadioControllerImpl.kt
+++ b/core/service/src/androidMain/kotlin/org/meshtastic/core/service/AndroidRadioControllerImpl.kt
@@ -17,15 +17,22 @@
package org.meshtastic.core.service
import android.content.Context
+import android.content.Intent
import co.touchlab.kermit.Logger
import kotlinx.coroutines.flow.StateFlow
import org.koin.core.annotation.Single
import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.DataPacket
+import org.meshtastic.core.model.Position
import org.meshtastic.core.model.RadioController
import org.meshtastic.core.model.service.ServiceAction
import org.meshtastic.core.repository.NodeRepository
+import org.meshtastic.proto.Channel
import org.meshtastic.proto.ClientNotification
+import org.meshtastic.proto.Config
+import org.meshtastic.proto.ModuleConfig
+import org.meshtastic.proto.SharedContact
+import org.meshtastic.proto.User
/**
* Android [RadioController] implementation that delegates to the bound [MeshService] via AIDL.
@@ -69,41 +76,37 @@ class AndroidRadioControllerImpl(
override suspend fun sendSharedContact(nodeNum: Int): Boolean {
val nodeDef = nodeRepository.getNode(DataPacket.nodeNumToDefaultId(nodeNum))
val contact =
- org.meshtastic.proto.SharedContact(
- node_num = nodeDef.num,
- user = nodeDef.user,
- manually_verified = nodeDef.manuallyVerified,
- )
+ SharedContact(node_num = nodeDef.num, user = nodeDef.user, manually_verified = nodeDef.manuallyVerified)
val action = ServiceAction.SendContact(contact)
serviceRepository.onServiceAction(action)
return action.result.await()
}
- override suspend fun setLocalConfig(config: org.meshtastic.proto.Config) {
+ override suspend fun setLocalConfig(config: Config) {
serviceRepository.meshService?.setConfig(config.encode())
}
- override suspend fun setLocalChannel(channel: org.meshtastic.proto.Channel) {
+ override suspend fun setLocalChannel(channel: Channel) {
serviceRepository.meshService?.setChannel(channel.encode())
}
- override suspend fun setOwner(destNum: Int, user: org.meshtastic.proto.User, packetId: Int) {
+ override suspend fun setOwner(destNum: Int, user: User, packetId: Int) {
serviceRepository.meshService?.setRemoteOwner(packetId, destNum, user.encode())
}
- override suspend fun setConfig(destNum: Int, config: org.meshtastic.proto.Config, packetId: Int) {
+ override suspend fun setConfig(destNum: Int, config: Config, packetId: Int) {
serviceRepository.meshService?.setRemoteConfig(packetId, destNum, config.encode())
}
- override suspend fun setModuleConfig(destNum: Int, config: org.meshtastic.proto.ModuleConfig, packetId: Int) {
+ override suspend fun setModuleConfig(destNum: Int, config: ModuleConfig, packetId: Int) {
serviceRepository.meshService?.setModuleConfig(packetId, destNum, config.encode())
}
- override suspend fun setRemoteChannel(destNum: Int, channel: org.meshtastic.proto.Channel, packetId: Int) {
+ override suspend fun setRemoteChannel(destNum: Int, channel: Channel, packetId: Int) {
serviceRepository.meshService?.setRemoteChannel(packetId, destNum, channel.encode())
}
- override suspend fun setFixedPosition(destNum: Int, position: org.meshtastic.core.model.Position) {
+ override suspend fun setFixedPosition(destNum: Int, position: Position) {
serviceRepository.meshService?.setFixedPosition(destNum, position)
}
@@ -171,7 +174,7 @@ class AndroidRadioControllerImpl(
serviceRepository.meshService?.removeByNodenum(packetId, nodeNum)
}
- override suspend fun requestPosition(destNum: Int, currentPosition: org.meshtastic.core.model.Position) {
+ override suspend fun requestPosition(destNum: Int, currentPosition: Position) {
serviceRepository.meshService?.requestPosition(destNum, currentPosition)
}
@@ -214,10 +217,7 @@ class AndroidRadioControllerImpl(
@Suppress("DEPRECATION") // Internal use: routes address change through AIDL binder
serviceRepository.meshService?.setDeviceAddress(address)
// Ensure service is running/restarted to handle the new address
- val intent =
- android.content.Intent().apply {
- setClassName("com.geeksville.mesh", "org.meshtastic.core.service.MeshService")
- }
+ val intent = Intent().apply { setClassName("com.geeksville.mesh", "org.meshtastic.core.service.MeshService") }
context.startForegroundService(intent)
}
}
diff --git a/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshService.kt b/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshService.kt
index 05f1135f10..028030f765 100644
--- a/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshService.kt
+++ b/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshService.kt
@@ -50,6 +50,12 @@ import org.meshtastic.core.repository.ServiceBroadcasts
import org.meshtastic.core.repository.ServiceRepository
import org.meshtastic.proto.PortNum
+/**
+ * Android foreground service that hosts the Meshtastic mesh radio connection.
+ *
+ * Acts as the lifecycle anchor for the [MeshServiceOrchestrator], which manages all manager initialization and
+ * connection state. Exposes an AIDL binder for external client integration via [core:api].
+ */
// IMeshService is deprecated but still required for AIDL binding
@Suppress("TooManyFunctions", "LargeClass", "DEPRECATION")
class MeshService : Service() {
diff --git a/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshServiceNotificationsImpl.kt b/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshServiceNotificationsImpl.kt
index 75bbe27ce5..cff4ec0417 100644
--- a/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshServiceNotificationsImpl.kt
+++ b/core/service/src/androidMain/kotlin/org/meshtastic/core/service/MeshServiceNotificationsImpl.kt
@@ -41,6 +41,7 @@ import org.jetbrains.compose.resources.StringResource
import org.koin.core.annotation.Single
import org.meshtastic.core.common.util.NumberFormatter
import org.meshtastic.core.common.util.nowMillis
+import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.Message
import org.meshtastic.core.model.Node
@@ -303,17 +304,14 @@ class MeshServiceNotificationsImpl(
// region Public Notification Methods
@Suppress("CyclomaticComplexMethod", "NestedBlockDepth")
- override fun updateServiceStateNotification(
- state: org.meshtastic.core.model.ConnectionState,
- telemetry: Telemetry?,
- ) {
+ override fun updateServiceStateNotification(state: ConnectionState, telemetry: Telemetry?) {
val summaryString =
when (state) {
- is org.meshtastic.core.model.ConnectionState.Connected ->
+ is ConnectionState.Connected ->
getString(Res.string.meshtastic_app_name) + ": " + getString(Res.string.connected)
- is org.meshtastic.core.model.ConnectionState.Disconnected -> getString(Res.string.disconnected)
- is org.meshtastic.core.model.ConnectionState.DeviceSleep -> getString(Res.string.device_sleeping)
- is org.meshtastic.core.model.ConnectionState.Connecting -> getString(Res.string.connecting)
+ is ConnectionState.Disconnected -> getString(Res.string.disconnected)
+ is ConnectionState.DeviceSleep -> getString(Res.string.device_sleeping)
+ is ConnectionState.Connecting -> getString(Res.string.connecting)
}
// Update caches if telemetry is provided
diff --git a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
index 1865dd4c66..df860a4a23 100644
--- a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
+++ b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt
@@ -19,9 +19,12 @@ package org.meshtastic.core.service
import androidx.lifecycle.Lifecycle
import androidx.lifecycle.coroutineScope
import co.touchlab.kermit.Logger
+import kotlinx.atomicfu.atomic
import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
@@ -96,10 +99,7 @@ class SharedRadioInterfaceService(
override val receivedData: SharedFlow<ByteArray> = _receivedData
private val _meshActivity =
- MutableSharedFlow<MeshActivity>(
- extraBufferCapacity = 64,
- onBufferOverflow = kotlinx.coroutines.channels.BufferOverflow.DROP_OLDEST,
- )
+ MutableSharedFlow<MeshActivity>(extraBufferCapacity = 64, onBufferOverflow = BufferOverflow.DROP_OLDEST)
override val meshActivity: SharedFlow<MeshActivity> = _meshActivity.asSharedFlow()
private val _connectionError = MutableSharedFlow<String>(extraBufferCapacity = 64)
@@ -109,12 +109,12 @@ class SharedRadioInterfaceService(
get() = _serviceScope
private var _serviceScope = CoroutineScope(dispatchers.io + SupervisorJob())
- private var radioIf: RadioTransport? = null
- private var runningInterfaceId: InterfaceId? = null
+ private var radioTransport: RadioTransport? = null
+ private var runningTransportId: InterfaceId? = null
private var isStarted = false
- private val listenersInitialized = kotlinx.atomicfu.atomic(false)
- private var heartbeatJob: kotlinx.coroutines.Job? = null
+ private val listenersInitialized = atomic(false)
+ private var heartbeatJob: Job? = null
private var lastHeartbeatMillis = 0L
@Volatile private var lastDataReceivedMillis = 0L
@@ -130,7 +130,7 @@ class SharedRadioInterfaceService(
}
private val initLock = Mutex()
- private val interfaceMutex = Mutex()
+ private val transportMutex = Mutex()
private fun initStateListeners() {
if (listenersInitialized.value) return
@@ -141,10 +141,10 @@ class SharedRadioInterfaceService(
radioPrefs.devAddr
.onEach { addr ->
- interfaceMutex.withLock {
+ transportMutex.withLock {
if (_currentDeviceAddressFlow.value != addr) {
_currentDeviceAddressFlow.value = addr
- startInterfaceLocked()
+ startTransportLocked()
}
}
}
@@ -152,11 +152,11 @@ class SharedRadioInterfaceService(
bluetoothRepository.state
.onEach { state ->
- interfaceMutex.withLock {
+ transportMutex.withLock {
if (state.enabled) {
- startInterfaceLocked()
- } else if (runningInterfaceId == InterfaceId.BLUETOOTH) {
- stopInterfaceLocked()
+ startTransportLocked()
+ } else if (runningTransportId == InterfaceId.BLUETOOTH) {
+ stopTransportLocked()
}
}
}
@@ -165,11 +165,11 @@ class SharedRadioInterfaceService(
networkRepository.networkAvailable
.onEach { state ->
- interfaceMutex.withLock {
+ transportMutex.withLock {
if (state) {
- startInterfaceLocked()
- } else if (runningInterfaceId == InterfaceId.TCP) {
- stopInterfaceLocked()
+ startTransportLocked()
+ } else if (runningTransportId == InterfaceId.TCP) {
+ stopTransportLocked()
}
}
}
@@ -180,11 +180,11 @@ class SharedRadioInterfaceService(
}
override fun connect() {
- processLifecycle.coroutineScope.launch { interfaceMutex.withLock { startInterfaceLocked() } }
+ processLifecycle.coroutineScope.launch { transportMutex.withLock { startTransportLocked() } }
initStateListeners()
}
- override fun isMockInterface(): Boolean = transportFactory.isMockInterface()
+ override fun isMockTransport(): Boolean = transportFactory.isMockTransport()
override fun toInterfaceAddress(interfaceId: InterfaceId, rest: String): String =
transportFactory.toInterfaceAddress(interfaceId, rest)
@@ -215,17 +215,17 @@ class SharedRadioInterfaceService(
_currentDeviceAddressFlow.value = sanitized
processLifecycle.coroutineScope.launch {
- interfaceMutex.withLock {
- ignoreException { stopInterfaceLocked() }
- startInterfaceLocked()
+ transportMutex.withLock {
+ ignoreException { stopTransportLocked() }
+ startTransportLocked()
}
}
return true
}
- /** Must be called under [interfaceMutex]. */
- private fun startInterfaceLocked() {
- if (radioIf != null) return
+ /** Must be called under [transportMutex]. */
+ private fun startTransportLocked() {
+ if (radioTransport != null) return
// Never autoconnect to the simulated node. The mock transport may be offered in the
// device-picker UI on debug builds, but it must only connect when the user explicitly
@@ -237,26 +237,26 @@ class SharedRadioInterfaceService(
return
}
- Logger.i { "Starting radio interface for ${address.anonymize}" }
+ Logger.i { "Starting radio transport for ${address.anonymize}" }
isStarted = true
- runningInterfaceId = address.firstOrNull()?.let { InterfaceId.forIdChar(it) }
- radioIf = transportFactory.createTransport(address, this)
+ runningTransportId = address.firstOrNull()?.let { InterfaceId.forIdChar(it) }
+ radioTransport = transportFactory.createTransport(address, this)
startHeartbeat()
}
- /** Must be called under [interfaceMutex]. */
- private fun stopInterfaceLocked() {
- val currentIf = radioIf
- Logger.i { "Stopping interface $currentIf" }
+ /** Must be called under [transportMutex]. */
+ private fun stopTransportLocked() {
+ val currentTransport = radioTransport
+ Logger.i { "Stopping transport $currentTransport" }
isStarted = false
- radioIf = null
- runningInterfaceId = null
- currentIf?.close()
+ radioTransport = null
+ runningTransportId = null
+ currentTransport?.close()
- _serviceScope.cancel("stopping interface")
+ _serviceScope.cancel("stopping transport")
_serviceScope = CoroutineScope(dispatchers.io + SupervisorJob())
- if (currentIf != null) {
+ if (currentTransport != null) {
onDisconnect(isPermanent = true)
}
}
@@ -295,23 +295,25 @@ class SharedRadioInterfaceService(
fun keepAlive(now: Long = nowMillis) {
if (now - lastHeartbeatMillis > HEARTBEAT_INTERVAL_MILLIS) {
- radioIf?.keepAlive()
+ radioTransport?.keepAlive()
lastHeartbeatMillis = now
}
}
override fun sendToRadio(bytes: ByteArray) {
- // Capture radioIf reference atomically to avoid racing with stopInterfaceLocked()
- // which sets radioIf = null and cancels _serviceScope. Without this snapshot,
- // we could read a non-null radioIf but launch into an already-cancelled scope.
- val currentIf =
- radioIf
+ // Snapshot the transport to avoid calling handleSendToRadio on a null reference.
+ // There is still a benign race: stopTransportLocked() may cancel _serviceScope
+ // between the null-check and the launch, causing the coroutine to be silently
+ // dropped. This is acceptable β if the transport is shutting down, dropping the
+ // send is the correct behavior.
+ val currentTransport =
+ radioTransport
?: run {
- Logger.w { "sendToRadio: no active radio interface, dropping ${bytes.size} bytes" }
+ Logger.w { "sendToRadio: no active radio transport, dropping ${bytes.size} bytes" }
return
}
_serviceScope.handledLaunch {
- currentIf.handleSendToRadio(bytes)
+ currentTransport.handleSendToRadio(bytes)
_meshActivity.tryEmit(MeshActivity.Send)
}
}
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeMeshServiceNotifications.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeMeshServiceNotifications.kt
index dc36b9956b..4f0a4b1530 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeMeshServiceNotifications.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeMeshServiceNotifications.kt
@@ -16,6 +16,7 @@
*/
package org.meshtastic.core.testing
+import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.Node
import org.meshtastic.core.repository.MeshServiceNotifications
import org.meshtastic.proto.ClientNotification
@@ -28,10 +29,7 @@ class FakeMeshServiceNotifications : MeshServiceNotifications {
override fun initChannels() {}
- override fun updateServiceStateNotification(
- state: org.meshtastic.core.model.ConnectionState,
- telemetry: Telemetry?,
- ) {}
+ override fun updateServiceStateNotification(state: ConnectionState, telemetry: Telemetry?) {}
override suspend fun updateMessageNotification(
contactKey: String,
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt
index fac69e28cc..d23a7f1ec5 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt
@@ -19,8 +19,13 @@ package org.meshtastic.core.testing
import kotlinx.coroutines.flow.StateFlow
import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.DataPacket
+import org.meshtastic.core.model.Position
import org.meshtastic.core.model.RadioController
+import org.meshtastic.proto.Channel
import org.meshtastic.proto.ClientNotification
+import org.meshtastic.proto.Config
+import org.meshtastic.proto.ModuleConfig
+import org.meshtastic.proto.User
/**
* A test double for [RadioController] that provides a no-op implementation and tracks calls for assertions in tests.
@@ -79,19 +84,19 @@ class FakeRadioController :
return true
}
- override suspend fun setLocalConfig(config: org.meshtastic.proto.Config) {}
+ override suspend fun setLocalConfig(config: Config) {}
- override suspend fun setLocalChannel(channel: org.meshtastic.proto.Channel) {}
+ override suspend fun setLocalChannel(channel: Channel) {}
- override suspend fun setOwner(destNum: Int, user: org.meshtastic.proto.User, packetId: Int) {}
+ override suspend fun setOwner(destNum: Int, user: User, packetId: Int) {}
- override suspend fun setConfig(destNum: Int, config: org.meshtastic.proto.Config, packetId: Int) {}
+ override suspend fun setConfig(destNum: Int, config: Config, packetId: Int) {}
- override suspend fun setModuleConfig(destNum: Int, config: org.meshtastic.proto.ModuleConfig, packetId: Int) {}
+ override suspend fun setModuleConfig(destNum: Int, config: ModuleConfig, packetId: Int) {}
- override suspend fun setRemoteChannel(destNum: Int, channel: org.meshtastic.proto.Channel, packetId: Int) {}
+ override suspend fun setRemoteChannel(destNum: Int, channel: Channel, packetId: Int) {}
- override suspend fun setFixedPosition(destNum: Int, position: org.meshtastic.core.model.Position) {}
+ override suspend fun setFixedPosition(destNum: Int, position: Position) {}
override suspend fun setRingtone(destNum: Int, ringtone: String) {}
@@ -125,7 +130,7 @@ class FakeRadioController :
override suspend fun removeByNodenum(packetId: Int, nodeNum: Int) {}
- override suspend fun requestPosition(destNum: Int, currentPosition: org.meshtastic.core.model.Position) {}
+ override suspend fun requestPosition(destNum: Int, currentPosition: Position) {}
override suspend fun requestUserInfo(destNum: Int) {}
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioInterfaceService.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioInterfaceService.kt
index 3b8c83fe99..9f11a2bc62 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioInterfaceService.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioInterfaceService.kt
@@ -60,7 +60,7 @@ class FakeRadioInterfaceService(override val serviceScope: CoroutineScope = Main
val sentToRadio = mutableListOf<ByteArray>()
var connectCalled = false
- override fun isMockInterface(): Boolean = true
+ override fun isMockTransport(): Boolean = true
override fun sendToRadio(bytes: ByteArray) {
sentToRadio.add(bytes)
diff --git a/core/ui/src/commonMain/kotlin/org/meshtastic/core/ui/viewmodel/UIViewModel.kt b/core/ui/src/commonMain/kotlin/org/meshtastic/core/ui/viewmodel/UIViewModel.kt
index 1e20213046..b1c4cebf28 100644
--- a/core/ui/src/commonMain/kotlin/org/meshtastic/core/ui/viewmodel/UIViewModel.kt
+++ b/core/ui/src/commonMain/kotlin/org/meshtastic/core/ui/viewmodel/UIViewModel.kt
@@ -18,6 +18,7 @@ package org.meshtastic.core.ui.viewmodel
import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
+import androidx.navigation3.runtime.NavKey
import co.touchlab.kermit.Logger
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.Flow
@@ -34,6 +35,7 @@ import kotlinx.coroutines.flow.onEach
import org.jetbrains.compose.resources.StringResource
import org.jetbrains.compose.resources.getString
import org.koin.core.annotation.KoinViewModel
+import org.meshtastic.core.common.util.CommonUri
import org.meshtastic.core.common.util.MeshtasticUri
import org.meshtastic.core.database.entity.asDeviceVersion
import org.meshtastic.core.model.MeshActivity
@@ -43,6 +45,7 @@ import org.meshtastic.core.model.TracerouteMapAvailability
import org.meshtastic.core.model.evaluateTracerouteMapAvailability
import org.meshtastic.core.model.service.TracerouteResponse
import org.meshtastic.core.model.util.dispatchMeshtasticUri
+import org.meshtastic.core.navigation.DeepLinkRouter
import org.meshtastic.core.repository.FirmwareReleaseRepository
import org.meshtastic.core.repository.MeshLogRepository
import org.meshtastic.core.repository.NodeRepository
@@ -84,7 +87,7 @@ class UIViewModel(
val snackbarManager: SnackbarManager,
) : ViewModel() {
- private val _navigationDeepLink = MutableSharedFlow<List<androidx.navigation3.runtime.NavKey>>(replay = 1)
+ private val _navigationDeepLink = MutableSharedFlow<List<NavKey>>(replay = 1)
val navigationDeepLink = _navigationDeepLink.asSharedFlow()
/**
@@ -97,10 +100,10 @@ class UIViewModel(
* [dispatchMeshtasticUri]. This triggers import dialogs for shared nodes or channel configurations.
*/
fun handleDeepLink(uri: MeshtasticUri, onInvalid: () -> Unit = {}) {
- val commonUri = org.meshtastic.core.common.util.CommonUri.parse(uri.uriString)
+ val commonUri = CommonUri.parse(uri.uriString)
// Try navigation routing first
- val navKeys = org.meshtastic.core.navigation.DeepLinkRouter.route(commonUri)
+ val navKeys = DeepLinkRouter.route(commonUri)
if (navKeys != null) {
_navigationDeepLink.tryEmit(navKeys)
return
@@ -236,7 +239,7 @@ class UIViewModel(
_sharedContactRequested.value = contact
}
- /** Called immediately after activity observes requestChannelUrl */
+ /** Clears the pending shared contact request. */
fun clearSharedContactRequested() {
_sharedContactRequested.value = null
}
@@ -255,7 +258,7 @@ class UIViewModel(
val latestStableFirmwareRelease = firmwareReleaseRepository.stableRelease.mapNotNull { it?.asDeviceVersion() }
- /** Called immediately after activity observes requestChannelUrl */
+ /** Clears the pending channel set import request. */
fun clearRequestChannelUrl() {
_requestChannelSet.value = null
}
diff --git a/desktop/src/main/kotlin/org/meshtastic/desktop/di/DesktopKoinModule.kt b/desktop/src/main/kotlin/org/meshtastic/desktop/di/DesktopKoinModule.kt
index 978be6b263..336f87b549 100644
--- a/desktop/src/main/kotlin/org/meshtastic/desktop/di/DesktopKoinModule.kt
+++ b/desktop/src/main/kotlin/org/meshtastic/desktop/di/DesktopKoinModule.kt
@@ -34,16 +34,25 @@ import org.meshtastic.core.model.NetworkFirmwareReleases
import org.meshtastic.core.model.RadioController
import org.meshtastic.core.network.KermitHttpLogger
import org.meshtastic.core.network.repository.MQTTRepository
+import org.meshtastic.core.network.service.ApiService
+import org.meshtastic.core.network.service.ApiServiceImpl
import org.meshtastic.core.repository.AppWidgetUpdater
import org.meshtastic.core.repository.LocationRepository
import org.meshtastic.core.repository.MeshLocationManager
import org.meshtastic.core.repository.MeshServiceNotifications
import org.meshtastic.core.repository.MeshWorkerManager
import org.meshtastic.core.repository.MessageQueue
+import org.meshtastic.core.repository.NotificationManager
import org.meshtastic.core.repository.PlatformAnalytics
import org.meshtastic.core.repository.RadioTransportFactory
import org.meshtastic.core.repository.ServiceBroadcasts
import org.meshtastic.core.repository.ServiceRepository
+import org.meshtastic.core.service.DirectRadioControllerImpl
+import org.meshtastic.core.service.ServiceRepositoryImpl
+import org.meshtastic.desktop.DesktopBuildConfig
+import org.meshtastic.desktop.DesktopNotificationManager
+import org.meshtastic.desktop.notification.DesktopMeshServiceNotifications
+import org.meshtastic.desktop.radio.DesktopMessageQueue
import org.meshtastic.desktop.radio.DesktopRadioTransportFactory
import org.meshtastic.desktop.stub.NoopAppWidgetUpdater
import org.meshtastic.desktop.stub.NoopCompassHeadingProvider
@@ -55,6 +64,9 @@ import org.meshtastic.desktop.stub.NoopMeshWorkerManager
import org.meshtastic.desktop.stub.NoopPhoneLocationProvider
import org.meshtastic.desktop.stub.NoopPlatformAnalytics
import org.meshtastic.desktop.stub.NoopServiceBroadcasts
+import org.meshtastic.feature.node.compass.CompassHeadingProvider
+import org.meshtastic.feature.node.compass.MagneticFieldProvider
+import org.meshtastic.feature.node.compass.PhoneLocationProvider
import org.meshtastic.core.ble.di.module as coreBleModule
import org.meshtastic.core.common.di.module as coreCommonModule
import org.meshtastic.core.data.di.module as coreDataModule
@@ -124,7 +136,7 @@ fun desktopModule() = module {
*/
@Suppress("LongMethod")
private fun desktopPlatformStubsModule() = module {
- single<ServiceRepository> { org.meshtastic.core.service.ServiceRepositoryImpl() }
+ single<ServiceRepository> { ServiceRepositoryImpl() }
single<RadioTransportFactory> {
DesktopRadioTransportFactory(
dispatchers = get(),
@@ -134,7 +146,7 @@ private fun desktopPlatformStubsModule() = module {
)
}
single<RadioController> {
- org.meshtastic.core.service.DirectRadioControllerImpl(
+ DirectRadioControllerImpl(
serviceRepository = get(),
nodeRepository = get(),
commandSender = get(),
@@ -144,37 +156,29 @@ private fun desktopPlatformStubsModule() = module {
locationManager = get(),
)
}
- single { org.meshtastic.desktop.DesktopNotificationManager(prefs = get()) }
- single<org.meshtastic.core.repository.NotificationManager> {
- get<org.meshtastic.desktop.DesktopNotificationManager>()
- }
- single<MeshServiceNotifications> {
- org.meshtastic.desktop.notification.DesktopMeshServiceNotifications(notificationManager = get())
- }
+ single { DesktopNotificationManager(prefs = get()) }
+ single<NotificationManager> { get<DesktopNotificationManager>() }
+ single<MeshServiceNotifications> { DesktopMeshServiceNotifications(notificationManager = get()) }
single<PlatformAnalytics> { NoopPlatformAnalytics() }
single<ServiceBroadcasts> { NoopServiceBroadcasts() }
single<AppWidgetUpdater> { NoopAppWidgetUpdater() }
single<MeshWorkerManager> { NoopMeshWorkerManager() }
- single<MessageQueue> {
- org.meshtastic.desktop.radio.DesktopMessageQueue(packetRepository = get(), radioController = get())
- }
+ single<MessageQueue> { DesktopMessageQueue(packetRepository = get(), radioController = get()) }
single<MeshLocationManager> { NoopMeshLocationManager() }
single<LocationRepository> { NoopLocationRepository() }
single<MQTTRepository> { NoopMQTTRepository() }
- single<org.meshtastic.feature.node.compass.CompassHeadingProvider> { NoopCompassHeadingProvider() }
- single<org.meshtastic.feature.node.compass.PhoneLocationProvider> { NoopPhoneLocationProvider() }
- single<org.meshtastic.feature.node.compass.MagneticFieldProvider> { NoopMagneticFieldProvider() }
+ single<CompassHeadingProvider> { NoopCompassHeadingProvider() }
+ single<PhoneLocationProvider> { NoopPhoneLocationProvider() }
+ single<MagneticFieldProvider> { NoopMagneticFieldProvider() }
// Desktop uses the real ApiService implementation (no flavor stub needed)
- single<org.meshtastic.core.network.service.ApiService> {
- org.meshtastic.core.network.service.ApiServiceImpl(client = get())
- }
+ single<ApiService> { ApiServiceImpl(client = get()) }
// Ktor HttpClient for JVM/Desktop (equivalent of CoreNetworkAndroidModule on Android)
single<HttpClient> {
HttpClient(Java) {
install(ContentNegotiation) { json(get<Json>()) }
- if (org.meshtastic.desktop.DesktopBuildConfig.IS_DEBUG) {
+ if (DesktopBuildConfig.IS_DEBUG) {
install(Logging) {
logger = KermitHttpLogger
level = LogLevel.HEADERS
diff --git a/desktop/src/main/kotlin/org/meshtastic/desktop/radio/DesktopRadioTransportFactory.kt b/desktop/src/main/kotlin/org/meshtastic/desktop/radio/DesktopRadioTransportFactory.kt
index 0518620c06..ffaa0553b5 100644
--- a/desktop/src/main/kotlin/org/meshtastic/desktop/radio/DesktopRadioTransportFactory.kt
+++ b/desktop/src/main/kotlin/org/meshtastic/desktop/radio/DesktopRadioTransportFactory.kt
@@ -24,7 +24,7 @@ import org.meshtastic.core.model.DeviceType
import org.meshtastic.core.model.InterfaceId
import org.meshtastic.core.network.SerialTransport
import org.meshtastic.core.network.radio.BaseRadioTransportFactory
-import org.meshtastic.core.network.radio.TCPInterface
+import org.meshtastic.core.network.radio.TcpRadioTransport
import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.core.repository.RadioTransport
import org.meshtastic.core.repository.RadioTransportFactory
@@ -45,16 +45,22 @@ class DesktopRadioTransportFactory(
override val supportedDeviceTypes: List<DeviceType> = listOf(DeviceType.TCP, DeviceType.BLE, DeviceType.USB)
- override fun isMockInterface(): Boolean = false
+ override fun isMockTransport(): Boolean = false
override fun createPlatformTransport(address: String, service: RadioInterfaceService): RadioTransport = when {
address.startsWith(InterfaceId.TCP.id) -> {
- TCPInterface(service, dispatchers, address.removePrefix(InterfaceId.TCP.id.toString()))
+ TcpRadioTransport(
+ callback = service,
+ scope = service.serviceScope,
+ dispatchers = dispatchers,
+ address = address.removePrefix(InterfaceId.TCP.id.toString()),
+ )
}
address.startsWith(InterfaceId.SERIAL.id) -> {
SerialTransport.open(
portName = address.removePrefix(InterfaceId.SERIAL.id.toString()),
- service = service,
+ callback = service,
+ scope = service.serviceScope,
dispatchers = dispatchers,
)
}
diff --git a/desktop/src/main/kotlin/org/meshtastic/desktop/stub/NoopStubs.kt b/desktop/src/main/kotlin/org/meshtastic/desktop/stub/NoopStubs.kt
index 8d53990e21..220b21d058 100644
--- a/desktop/src/main/kotlin/org/meshtastic/desktop/stub/NoopStubs.kt
+++ b/desktop/src/main/kotlin/org/meshtastic/desktop/stub/NoopStubs.kt
@@ -20,6 +20,7 @@ package org.meshtastic.desktop.stub
import co.touchlab.kermit.Logger
import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
@@ -27,6 +28,7 @@ import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.emptyFlow
import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.DataPacket
+import org.meshtastic.core.model.DeviceType
import org.meshtastic.core.model.InterfaceId
import org.meshtastic.core.model.MeshActivity
import org.meshtastic.core.model.MessageStatus
@@ -37,14 +39,11 @@ import org.meshtastic.core.repository.DataPair
import org.meshtastic.core.repository.Location
import org.meshtastic.core.repository.LocationRepository
import org.meshtastic.core.repository.MeshLocationManager
-import org.meshtastic.core.repository.MeshServiceNotifications
import org.meshtastic.core.repository.MeshWorkerManager
import org.meshtastic.core.repository.PlatformAnalytics
import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.core.repository.ServiceBroadcasts
-import org.meshtastic.proto.ClientNotification
import org.meshtastic.proto.MqttClientProxyMessage
-import org.meshtastic.proto.Telemetry
import org.meshtastic.proto.Position as ProtoPosition
/**
@@ -66,12 +65,12 @@ private fun logWarn(message: String) {
// region Transport / Radio Stubs (Android BLE/USB β no commonMain impl)
class NoopRadioInterfaceService : RadioInterfaceService {
- override val supportedDeviceTypes: List<org.meshtastic.core.model.DeviceType> = emptyList()
+ override val supportedDeviceTypes: List<DeviceType> = emptyList()
override val connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected)
override val currentDeviceAddressFlow = MutableStateFlow<String?>(null)
- override fun isMockInterface(): Boolean = false
+ override fun isMockTransport(): Boolean = false
override val receivedData = MutableSharedFlow<ByteArray>()
override val meshActivity = MutableSharedFlow<MeshActivity>()
@@ -98,65 +97,13 @@ class NoopRadioInterfaceService : RadioInterfaceService {
override fun handleFromRadio(bytes: ByteArray) {}
@Suppress("InjectDispatcher")
- override val serviceScope: CoroutineScope = CoroutineScope(SupervisorJob() + kotlinx.coroutines.Dispatchers.Default)
+ override val serviceScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
}
// endregion
// region Notification / Platform Stubs (Android-only)
-@Suppress("TooManyFunctions")
-class NoopMeshServiceNotifications : MeshServiceNotifications {
- override fun clearNotifications() {}
-
- override fun initChannels() {}
-
- override fun updateServiceStateNotification(
- state: org.meshtastic.core.model.ConnectionState,
- telemetry: Telemetry?,
- ) {}
-
- override suspend fun updateMessageNotification(
- contactKey: String,
- name: String,
- message: String,
- isBroadcast: Boolean,
- channelName: String?,
- isSilent: Boolean,
- ) {}
-
- override suspend fun updateWaypointNotification(
- contactKey: String,
- name: String,
- message: String,
- waypointId: Int,
- isSilent: Boolean,
- ) {}
-
- override suspend fun updateReactionNotification(
- contactKey: String,
- name: String,
- emoji: String,
- isBroadcast: Boolean,
- channelName: String?,
- isSilent: Boolean,
- ) {}
-
- override fun showAlertNotification(contactKey: String, name: String, alert: String) {}
-
- override fun showNewNodeSeenNotification(node: Node) {}
-
- override fun showOrUpdateLowBatteryNotification(node: Node, isRemote: Boolean) {}
-
- override fun showClientNotification(clientNotification: ClientNotification) {}
-
- override fun cancelMessageNotification(contactKey: String) {}
-
- override fun cancelLowBatteryNotification(node: Node) {}
-
- override fun clearClientNotification(notification: ClientNotification) {}
-}
-
class NoopPlatformAnalytics : PlatformAnalytics {
override fun track(event: String, vararg properties: DataPair) {}
diff --git a/docs/decisions/architecture-review-2026-03.md b/docs/decisions/architecture-review-2026-03.md
index 3d09d68f3a..68ed448099 100644
--- a/docs/decisions/architecture-review-2026-03.md
+++ b/docs/decisions/architecture-review-2026-03.md
@@ -60,7 +60,7 @@ The core transport abstraction was previously locked in `app/repository/radio/`
1. Defined `RadioTransport` interface in `core:repository/commonMain` (replacing `IRadioInterface`)
2. Moved `StreamFrameCodec`-based framing to `core:network/commonMain`
3. Moved TCP transport to `core:network/jvmAndroidMain`
-4. The remaining `app/repository/radio/` implementations (BLE, Serial, Mock) now implement `RadioTransport`.
+4. BLE, Serial, and Mock transports now reside in `core:network` and implement `RadioTransport`.
**Recommended next steps:**
1. Move BLE transport to `core:ble/androidMain`
diff --git a/docs/kmp-status.md b/docs/kmp-status.md
index 95e4b6945c..fb9d74175d 100644
--- a/docs/kmp-status.md
+++ b/docs/kmp-status.md
@@ -27,7 +27,7 @@ Modules that share JVM-specific code between Android and desktop now standardize
| `core:database` | β
| β
| Room KMP |
| `core:domain` | β
| β
| UseCases |
| `core:prefs` | β
| β
| Preferences layer |
-| `core:network` | β
| β
| Ktor, `StreamFrameCodec`, `TcpTransport`, `SerialTransport`, `BleRadioInterface` |
+| `core:network` | β
| β
| Ktor, `StreamFrameCodec`, `TcpTransport`, `SerialTransport`, `BleRadioTransport` |
| `core:data` | β
| β
| Data orchestration |
| `core:ble` | β
| β
| Kable multiplatform BLE abstractions in commonMain |
| `core:nfc` | β
| β
| NFC contract in commonMain; hardware in androidMain |
@@ -116,7 +116,7 @@ Based on the latest codebase investigation, the following steps are proposed to
| **Transport Lifecycle Unification** | β
Done | `SharedRadioInterfaceService` orchestrates auto-reconnect, connection state, and heartbeat uniformly across Android and Desktop. |
| **Database Parity** | β
Done | `DatabaseManager` is pure KMP, giving iOS and Desktop support for multiple connected nodes with LRU caching. On JVM/Desktop, inactive databases are explicitly closed on switch (Room KMP's `setAutoCloseTimeout` is Android-only), and `desktopDataDir()` in `core:database/jvmMain` is the single source for data directory resolution. |
| Emoji picker unification | β
Done | Single commonMain implementation replacing 3 platform variants |
-| Cross-platform deduplication pass | β
Done | Extracted shared `AlertHost`, `SharedDialogs`, `PlaceholderScreen`, `ThemePickerDialog`, `MeshtasticNavDisplay`, `formatLogsTo()`, `handleNodeAction()`, `findNodeByNameSuffix()`, `MeshtasticAppShell`, `BleRadioInterface`, and `BaseRadioTransportFactory` to `commonMain`; eliminated ~1,200 lines of duplicated Compose UI code across Android/desktop |
+| Cross-platform deduplication pass | β
Done | Extracted shared `AlertHost`, `SharedDialogs`, `PlaceholderScreen`, `ThemePickerDialog`, `MeshtasticNavDisplay`, `formatLogsTo()`, `handleNodeAction()`, `findNodeByNameSuffix()`, `MeshtasticAppShell`, `BleRadioTransport`, and `BaseRadioTransportFactory` to `commonMain`; eliminated ~1,200 lines of duplicated Compose UI code across Android/desktop |
## Navigation Parity Note
@@ -150,7 +150,7 @@ Extracted to shared `commonMain` (no longer app-only):
Extracted to core KMP modules:
- Android Services, WorkManager Workers, and BroadcastReceivers β `core:service/androidMain`
- USB/Serial radio connections β `core:network/androidMain`
-- TCP radio connections, BLE radio connections (`BleRadioInterface`), and mDNS/NSD Service Discovery β `core:network/commonMain` (with Android `NsdManager` and Desktop `JmDNS` implementations)
+- TCP radio connections, BLE radio connections (`BleRadioTransport`), and mDNS/NSD Service Discovery β `core:network/commonMain` (with Android `NsdManager` and Desktop `JmDNS` implementations)
Remaining to be extracted from `:app` or unified in `commonMain`:
- `MapViewModel` (Unify Google/F-Droid flavors into a single `commonMain` class consuming a `MapConfigProvider` interface. `MapViewProvider` interface simplified β track rendering and traceroute rendering extracted to dedicated provider contracts)
diff --git a/docs/roadmap.md b/docs/roadmap.md
index 91d051f9fd..9c94454856 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -57,7 +57,7 @@ These items address structural gaps identified in the March 2026 architecture re
| TCP | Desktop (JVM) | β
Done β shared `StreamFrameCodec` + `TcpTransport` in `core:network` |
| Serial/USB | Desktop (JVM) | β
Done β jSerialComm |
| MQTT | All (KMP) | β
Completed β KMQTT in commonMain |
-| BLE | All (KMP) | β
Done β Kable in `commonMain` (`BleRadioInterface`) |
+| BLE | All (KMP) | β
Done β Kable in `commonMain` (`BleRadioTransport`) |
### Desktop Feature Gaps
diff --git a/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ScannerViewModel.kt b/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ScannerViewModel.kt
index e4bb00c6b9..d094aa1704 100644
--- a/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ScannerViewModel.kt
+++ b/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ScannerViewModel.kt
@@ -54,8 +54,8 @@ open class ScannerViewModel(
private val dispatchers: org.meshtastic.core.di.CoroutineDispatchers,
private val bleScanner: org.meshtastic.core.ble.BleScanner? = null,
) : ViewModel() {
- private val _showMockInterface = MutableStateFlow(false)
- val showMockInterface: StateFlow<Boolean> = _showMockInterface.asStateFlow()
+ private val _showMockTransport = MutableStateFlow(false)
+ val showMockTransport: StateFlow<Boolean> = _showMockTransport.asStateFlow()
private val _errorText = MutableStateFlow<String?>(null)
val errorText: StateFlow<String?> = _errorText.asStateFlow()
@@ -68,7 +68,7 @@ open class ScannerViewModel(
private var scanJob: kotlinx.coroutines.Job? = null
init {
- _showMockInterface.value = radioInterfaceService.isMockInterface()
+ _showMockTransport.value = radioInterfaceService.isMockTransport()
}
fun startBleScan() {
@@ -77,25 +77,26 @@ open class ScannerViewModel(
isBleScanningState.value = true
scannedBleDevices.value = emptyMap()
- scanJob = viewModelScope.launch {
- try {
- bleScanner
- .scan(
- timeout = kotlin.time.Duration.INFINITE,
- serviceUuid = org.meshtastic.core.ble.MeshtasticBleConstants.SERVICE_UUID,
- )
- .flowOn(dispatchers.io)
- .collect { device ->
- if (!scannedBleDevices.value.containsKey(device.address)) {
- scannedBleDevices.update { current -> current + (device.address to device) }
+ scanJob =
+ viewModelScope.launch {
+ try {
+ bleScanner
+ .scan(
+ timeout = kotlin.time.Duration.INFINITE,
+ serviceUuid = org.meshtastic.core.ble.MeshtasticBleConstants.SERVICE_UUID,
+ )
+ .flowOn(dispatchers.io)
+ .collect { device ->
+ if (!scannedBleDevices.value.containsKey(device.address)) {
+ scannedBleDevices.update { current -> current + (device.address to device) }
+ }
}
- }
- } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
- co.touchlab.kermit.Logger.w(e) { "BLE scan failed" }
- } finally {
- isBleScanningState.value = false
+ } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
+ co.touchlab.kermit.Logger.w(e) { "BLE scan failed" }
+ } finally {
+ isBleScanningState.value = false
+ }
}
- }
}
fun stopBleScan() {
@@ -105,7 +106,7 @@ open class ScannerViewModel(
}
private val discoveredDevicesFlow =
- showMockInterface
+ showMockTransport
.flatMapLatest { showMock -> getDiscoveredDevicesUseCase.invoke(showMock) }
.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), null)
diff --git a/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/ConnectionsScreen.kt b/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/ConnectionsScreen.kt
index 441b81c84c..7fdc287cdd 100644
--- a/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/ConnectionsScreen.kt
+++ b/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/ConnectionsScreen.kt
@@ -167,17 +167,19 @@ fun ConnectionsScreen(
Spacer(modifier = Modifier.height(4.dp))
val uiState =
when {
- connectionState is ConnectionState.Connected && ourNode != null -> 2
+ connectionState is ConnectionState.Connected && ourNode != null ->
+ ConnectionUiState.CONNECTED_WITH_NODE
+
connectionState is ConnectionState.Connected ||
connectionState == ConnectionState.Connecting ||
- selectedDevice != NO_DEVICE_SELECTED -> 1
+ selectedDevice != NO_DEVICE_SELECTED -> ConnectionUiState.CONNECTING
- else -> 0
+ else -> ConnectionUiState.NO_DEVICE
}
Crossfade(targetState = uiState, label = "connection_state") { state ->
when (state) {
- 2 ->
+ ConnectionUiState.CONNECTED_WITH_NODE ->
ConnectedDeviceContent(
ourNode = ourNode,
regionUnset = regionUnset,
@@ -191,7 +193,7 @@ fun ConnectionsScreen(
},
)
- 1 ->
+ ConnectionUiState.CONNECTING ->
ConnectingDeviceContent(
connectionState = connectionState,
selectedDevice = selectedDevice,
@@ -208,7 +210,9 @@ fun ConnectionsScreen(
}
var selectedDeviceType by remember { mutableStateOf(DeviceType.BLE) }
- LaunchedEffect(Unit) { DeviceType.fromAddress(selectedDevice)?.let { selectedDeviceType = it } }
+ LaunchedEffect(selectedDevice) {
+ DeviceType.fromAddress(selectedDevice)?.let { selectedDeviceType = it }
+ }
val supportedDeviceTypes = scanModel.supportedDeviceTypes
@@ -369,3 +373,15 @@ private fun NoDeviceContent() {
)
}
}
+
+/** Visual state for the connection screen's [Crossfade] animation. */
+private enum class ConnectionUiState {
+ /** No device is selected. */
+ NO_DEVICE,
+
+ /** A device is selected or we are actively connecting. */
+ CONNECTING,
+
+ /** Connected with node info available. */
+ CONNECTED_WITH_NODE,
+}
diff --git a/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/components/ConnectingDeviceInfo.kt b/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/components/ConnectingDeviceInfo.kt
index 53cec80b57..ebc981398c 100644
--- a/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/components/ConnectingDeviceInfo.kt
+++ b/feature/connections/src/commonMain/kotlin/org/meshtastic/feature/connections/ui/components/ConnectingDeviceInfo.kt
@@ -42,6 +42,10 @@ import org.meshtastic.core.resources.connecting
import org.meshtastic.core.resources.disconnect
import org.meshtastic.core.ui.theme.StatusColors.StatusRed
+/**
+ * Displays the currently connecting (or connected) device with its name, address, connection status, and a disconnect
+ * button.
+ */
@Composable
fun ConnectingDeviceInfo(
connectionState: ConnectionState,
diff --git a/feature/connections/src/commonTest/kotlin/org/meshtastic/feature/connections/ScannerViewModelTest.kt b/feature/connections/src/commonTest/kotlin/org/meshtastic/feature/connections/ScannerViewModelTest.kt
index 6f291d68a9..04e9ac03ef 100644
--- a/feature/connections/src/commonTest/kotlin/org/meshtastic/feature/connections/ScannerViewModelTest.kt
+++ b/feature/connections/src/commonTest/kotlin/org/meshtastic/feature/connections/ScannerViewModelTest.kt
@@ -53,7 +53,7 @@ class ScannerViewModelTest {
@BeforeTest
fun setUp() {
- every { radioInterfaceService.isMockInterface() } returns false
+ every { radioInterfaceService.isMockTransport() } returns false
every { radioInterfaceService.currentDeviceAddressFlow } returns MutableStateFlow(null)
every { radioInterfaceService.supportedDeviceTypes } returns emptyList()
Served by rngit 1.5.0 - Generated in 0.9s